You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by td...@apache.org on 2015/11/24 06:24:27 UTC
[6/9] phoenix git commit: PHOENIX-1674 Snapshot isolation transaction
support through Tephra (James Taylor, Thomas D'Silva)
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
index 54c688a..65a43de 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ScanRegionObserver.java
@@ -26,7 +26,7 @@ import java.sql.SQLException;
import java.util.List;
import java.util.Set;
-import com.google.common.collect.Sets;
+import co.cask.tephra.Transaction;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HRegionInfo;
@@ -39,6 +39,7 @@ import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.cache.GlobalCache;
import org.apache.phoenix.cache.TenantCache;
+import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.execute.TupleProjector;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
@@ -58,8 +59,10 @@ import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
+import org.apache.phoenix.util.TransactionUtil;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
/**
@@ -191,6 +194,7 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
Region dataRegion = null;
IndexMaintainer indexMaintainer = null;
byte[][] viewConstants = null;
+ Transaction tx = null;
ColumnReference[] dataColumns = IndexUtil.deserializeDataTableColumnsToJoin(scan);
if (dataColumns != null) {
tupleProjector = IndexUtil.getTupleProjector(scan, dataColumns);
@@ -199,14 +203,16 @@ public class ScanRegionObserver extends BaseScannerRegionObserver {
List<IndexMaintainer> indexMaintainers = localIndexBytes == null ? null : IndexMaintainer.deserialize(localIndexBytes);
indexMaintainer = indexMaintainers.get(0);
viewConstants = IndexUtil.deserializeViewConstantsFromScan(scan);
+ byte[] txState = scan.getAttribute(BaseScannerRegionObserver.TX_STATE);
+ tx = MutationState.decodeTransaction(txState);
}
final TupleProjector p = TupleProjector.deserializeProjectorFromScan(scan);
final HashJoinInfo j = HashJoinInfo.deserializeHashJoinFromScan(scan);
innerScanner =
getWrappedScanner(c, innerScanner, arrayKVRefs, arrayFuncRefs, offset, scan,
- dataColumns, tupleProjector, dataRegion, indexMaintainer, viewConstants,
- kvSchema, kvSchemaBitSet, j == null ? p : null, ptr);
+ dataColumns, tupleProjector, dataRegion, indexMaintainer, tx,
+ viewConstants, kvSchema, kvSchemaBitSet, j == null ? p : null, ptr);
final ImmutableBytesWritable tenantId = ScanUtil.getTenantId(scan);
if (j != null) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
index 2f31b08..9f3bdb4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingEndpointImpl.java
@@ -26,18 +26,17 @@ import org.apache.hadoop.hbase.coprocessor.CoprocessorException;
import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-
import org.apache.phoenix.cache.GlobalCache;
import org.apache.phoenix.cache.TenantCache;
-import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
-
import org.apache.phoenix.coprocessor.ServerCachingProtocol.ServerCacheFactory;
import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.AddServerCacheRequest;
import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.AddServerCacheResponse;
import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.RemoveServerCacheRequest;
import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.RemoveServerCacheResponse;
import org.apache.phoenix.coprocessor.generated.ServerCachingProtos.ServerCachingService;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.protobuf.ProtobufUtil;
+import org.apache.phoenix.util.ByteUtil;
import com.google.protobuf.RpcCallback;
import com.google.protobuf.RpcController;
@@ -66,6 +65,7 @@ public class ServerCachingEndpointImpl extends ServerCachingService implements C
ImmutableBytesWritable cachePtr =
org.apache.phoenix.protobuf.ProtobufUtil
.toImmutableBytesWritable(request.getCachePtr());
+ byte[] txState = request.hasTxState() ? request.getTxState().toByteArray() : ByteUtil.EMPTY_BYTE_ARRAY;
try {
@SuppressWarnings("unchecked")
@@ -73,7 +73,7 @@ public class ServerCachingEndpointImpl extends ServerCachingService implements C
(Class<ServerCacheFactory>) Class.forName(request.getCacheFactory().getClassName());
ServerCacheFactory cacheFactory = serverCacheFactoryClass.newInstance();
tenantCache.addServerCache(new ImmutableBytesPtr(request.getCacheId().toByteArray()),
- cachePtr, cacheFactory);
+ cachePtr, txState, cacheFactory);
} catch (Throwable e) {
ProtobufUtil.setControllerException(controller, new IOException(e));
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
index 4fdfe99..b201c8e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/ServerCachingProtocol.java
@@ -36,18 +36,19 @@ import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
*/
public interface ServerCachingProtocol {
public static interface ServerCacheFactory extends Writable {
- public Closeable newCache(ImmutableBytesWritable cachePtr, MemoryChunk chunk) throws SQLException;
+ public Closeable newCache(ImmutableBytesWritable cachePtr, byte[] txState, MemoryChunk chunk) throws SQLException;
}
/**
* Add the cache to the region server cache.
* @param tenantId the tenantId or null if not applicable
* @param cacheId unique identifier of the cache
* @param cachePtr pointer to the byte array of the cache
+ * @param txState TODO
* @param cacheFactory factory that converts from byte array to object representation on the server side
* @return true on success and otherwise throws
* @throws SQLException
*/
- public boolean addServerCache(byte[] tenantId, byte[] cacheId, ImmutableBytesWritable cachePtr, ServerCacheFactory cacheFactory) throws SQLException;
+ public boolean addServerCache(byte[] tenantId, byte[] cacheId, ImmutableBytesWritable cachePtr, byte[] txState, ServerCacheFactory cacheFactory) throws SQLException;
/**
* Remove the cache from the region server cache. Called upon completion of
* the operation when cache is no longer needed.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 5133cb2..ffdc05f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -105,6 +105,8 @@ import org.apache.phoenix.util.TimeKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import co.cask.tephra.TxConstants;
+
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -403,6 +405,8 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver{
Delete delete = new Delete(firstKV.getRowArray(),
firstKV.getRowOffset(), firstKV.getRowLength(),ts);
mutations.add(delete);
+ // force tephra to ignore this deletes
+ delete.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]);
} else if (isUpsert) {
Arrays.fill(values, null);
int i = 0;
@@ -462,6 +466,8 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver{
results.get(0).getRowOffset(),
results.get(0).getRowLength());
delete.deleteColumns(deleteCF, deleteCQ, ts);
+ // force tephra to ignore this deletes
+ delete.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]);
mutations.add(delete);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
index 25f8271..7e71cd9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
@@ -3218,6 +3218,16 @@ public final class PTableProtos {
* <code>optional bool rowKeyOrderOptimizable = 26;</code>
*/
boolean getRowKeyOrderOptimizable();
+
+ // optional bool transactional = 27;
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ boolean hasTransactional();
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ boolean getTransactional();
}
/**
* Protobuf type {@code PTable}
@@ -3418,6 +3428,11 @@ public final class PTableProtos {
rowKeyOrderOptimizable_ = input.readBool();
break;
}
+ case 216: {
+ bitField0_ |= 0x00400000;
+ transactional_ = input.readBool();
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -3980,6 +3995,22 @@ public final class PTableProtos {
return rowKeyOrderOptimizable_;
}
+ // optional bool transactional = 27;
+ public static final int TRANSACTIONAL_FIELD_NUMBER = 27;
+ private boolean transactional_;
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ public boolean hasTransactional() {
+ return ((bitField0_ & 0x00400000) == 0x00400000);
+ }
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ public boolean getTransactional() {
+ return transactional_;
+ }
+
private void initFields() {
schemaNameBytes_ = com.google.protobuf.ByteString.EMPTY;
tableNameBytes_ = com.google.protobuf.ByteString.EMPTY;
@@ -4007,6 +4038,7 @@ public final class PTableProtos {
storeNulls_ = false;
baseColumnCount_ = 0;
rowKeyOrderOptimizable_ = false;
+ transactional_ = false;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -4152,6 +4184,9 @@ public final class PTableProtos {
if (((bitField0_ & 0x00200000) == 0x00200000)) {
output.writeBool(26, rowKeyOrderOptimizable_);
}
+ if (((bitField0_ & 0x00400000) == 0x00400000)) {
+ output.writeBool(27, transactional_);
+ }
getUnknownFields().writeTo(output);
}
@@ -4270,6 +4305,10 @@ public final class PTableProtos {
size += com.google.protobuf.CodedOutputStream
.computeBoolSize(26, rowKeyOrderOptimizable_);
}
+ if (((bitField0_ & 0x00400000) == 0x00400000)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBoolSize(27, transactional_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -4411,6 +4450,11 @@ public final class PTableProtos {
result = result && (getRowKeyOrderOptimizable()
== other.getRowKeyOrderOptimizable());
}
+ result = result && (hasTransactional() == other.hasTransactional());
+ if (hasTransactional()) {
+ result = result && (getTransactional()
+ == other.getTransactional());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -4528,6 +4572,10 @@ public final class PTableProtos {
hash = (37 * hash) + ROWKEYORDEROPTIMIZABLE_FIELD_NUMBER;
hash = (53 * hash) + hashBoolean(getRowKeyOrderOptimizable());
}
+ if (hasTransactional()) {
+ hash = (37 * hash) + TRANSACTIONAL_FIELD_NUMBER;
+ hash = (53 * hash) + hashBoolean(getTransactional());
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -4704,6 +4752,8 @@ public final class PTableProtos {
bitField0_ = (bitField0_ & ~0x01000000);
rowKeyOrderOptimizable_ = false;
bitField0_ = (bitField0_ & ~0x02000000);
+ transactional_ = false;
+ bitField0_ = (bitField0_ & ~0x04000000);
return this;
}
@@ -4852,6 +4902,10 @@ public final class PTableProtos {
to_bitField0_ |= 0x00200000;
}
result.rowKeyOrderOptimizable_ = rowKeyOrderOptimizable_;
+ if (((from_bitField0_ & 0x04000000) == 0x04000000)) {
+ to_bitField0_ |= 0x00400000;
+ }
+ result.transactional_ = transactional_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -5024,6 +5078,9 @@ public final class PTableProtos {
if (other.hasRowKeyOrderOptimizable()) {
setRowKeyOrderOptimizable(other.getRowKeyOrderOptimizable());
}
+ if (other.hasTransactional()) {
+ setTransactional(other.getTransactional());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -6694,6 +6751,39 @@ public final class PTableProtos {
return this;
}
+ // optional bool transactional = 27;
+ private boolean transactional_ ;
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ public boolean hasTransactional() {
+ return ((bitField0_ & 0x04000000) == 0x04000000);
+ }
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ public boolean getTransactional() {
+ return transactional_;
+ }
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ public Builder setTransactional(boolean value) {
+ bitField0_ |= 0x04000000;
+ transactional_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional bool transactional = 27;</code>
+ */
+ public Builder clearTransactional() {
+ bitField0_ = (bitField0_ & ~0x04000000);
+ transactional_ = false;
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:PTable)
}
@@ -6740,7 +6830,7 @@ public final class PTableProtos {
"leStats\022\013\n\003key\030\001 \002(\014\022\016\n\006values\030\002 \003(\014\022\033\n\023" +
"guidePostsByteCount\030\003 \001(\003\022\025\n\rkeyBytesCou",
"nt\030\004 \001(\003\022\027\n\017guidePostsCount\030\005 \001(\005\022!\n\013pGu" +
- "idePosts\030\006 \001(\0132\014.PGuidePosts\"\357\004\n\006PTable\022" +
+ "idePosts\030\006 \001(\0132\014.PGuidePosts\"\206\005\n\006PTable\022" +
"\027\n\017schemaNameBytes\030\001 \002(\014\022\026\n\016tableNameByt" +
"es\030\002 \002(\014\022\036\n\ttableType\030\003 \002(\0162\013.PTableType" +
"\022\022\n\nindexState\030\004 \001(\t\022\026\n\016sequenceNumber\030\005" +
@@ -6756,10 +6846,11 @@ public final class PTableProtos {
"wIndexId\030\025 \001(\005\022\021\n\tindexType\030\026 \001(\014\022\026\n\016sta" +
"tsTimeStamp\030\027 \001(\003\022\022\n\nstoreNulls\030\030 \001(\010\022\027\n" +
"\017baseColumnCount\030\031 \001(\005\022\036\n\026rowKeyOrderOpt" +
- "imizable\030\032 \001(\010*A\n\nPTableType\022\n\n\006SYSTEM\020\000" +
- "\022\010\n\004USER\020\001\022\010\n\004VIEW\020\002\022\t\n\005INDEX\020\003\022\010\n\004JOIN\020" +
- "\004B@\n(org.apache.phoenix.coprocessor.gene",
- "ratedB\014PTableProtosH\001\210\001\001\240\001\001"
+ "imizable\030\032 \001(\010\022\025\n\rtransactional\030\033 \001(\010*A\n" +
+ "\nPTableType\022\n\n\006SYSTEM\020\000\022\010\n\004USER\020\001\022\010\n\004VIE" +
+ "W\020\002\022\t\n\005INDEX\020\003\022\010\n\004JOIN\020\004B@\n(org.apache.p",
+ "hoenix.coprocessor.generatedB\014PTableProt" +
+ "osH\001\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -6783,7 +6874,7 @@ public final class PTableProtos {
internal_static_PTable_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_PTable_descriptor,
- new java.lang.String[] { "SchemaNameBytes", "TableNameBytes", "TableType", "IndexState", "SequenceNumber", "TimeStamp", "PkNameBytes", "BucketNum", "Columns", "Indexes", "IsImmutableRows", "GuidePosts", "DataTableNameBytes", "DefaultFamilyName", "DisableWAL", "MultiTenant", "ViewType", "ViewStatement", "PhysicalNames", "TenantId", "ViewIndexId", "IndexType", "StatsTimeStamp", "StoreNulls", "BaseColumnCount", "RowKeyOrderOptimizable", });
+ new java.lang.String[] { "SchemaNameBytes", "TableNameBytes", "TableType", "IndexState", "SequenceNumber", "TimeStamp", "PkNameBytes", "BucketNum", "Columns", "Indexes", "IsImmutableRows", "GuidePosts", "DataTableNameBytes", "DefaultFamilyName", "DisableWAL", "MultiTenant", "ViewType", "ViewStatement", "PhysicalNames", "TenantId", "ViewIndexId", "IndexType", "StatsTimeStamp", "StoreNulls", "BaseColumnCount", "RowKeyOrderOptimizable", "Transactional", });
return null;
}
};
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/ServerCachingProtos.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/ServerCachingProtos.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/ServerCachingProtos.java
index 69db21b..5ee1dfb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/ServerCachingProtos.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/ServerCachingProtos.java
@@ -699,6 +699,16 @@ public final class ServerCachingProtos {
* <code>required .ServerCacheFactory cacheFactory = 4;</code>
*/
org.apache.phoenix.coprocessor.generated.ServerCacheFactoryProtos.ServerCacheFactoryOrBuilder getCacheFactoryOrBuilder();
+
+ // optional bytes txState = 5;
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ boolean hasTxState();
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ com.google.protobuf.ByteString getTxState();
}
/**
* Protobuf type {@code AddServerCacheRequest}
@@ -787,6 +797,11 @@ public final class ServerCachingProtos {
bitField0_ |= 0x00000008;
break;
}
+ case 42: {
+ bitField0_ |= 0x00000010;
+ txState_ = input.readBytes();
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -903,11 +918,28 @@ public final class ServerCachingProtos {
return cacheFactory_;
}
+ // optional bytes txState = 5;
+ public static final int TXSTATE_FIELD_NUMBER = 5;
+ private com.google.protobuf.ByteString txState_;
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ public boolean hasTxState() {
+ return ((bitField0_ & 0x00000010) == 0x00000010);
+ }
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ public com.google.protobuf.ByteString getTxState() {
+ return txState_;
+ }
+
private void initFields() {
tenantId_ = com.google.protobuf.ByteString.EMPTY;
cacheId_ = com.google.protobuf.ByteString.EMPTY;
cachePtr_ = org.apache.phoenix.coprocessor.generated.ServerCachingProtos.ImmutableBytesWritable.getDefaultInstance();
cacheFactory_ = org.apache.phoenix.coprocessor.generated.ServerCacheFactoryProtos.ServerCacheFactory.getDefaultInstance();
+ txState_ = com.google.protobuf.ByteString.EMPTY;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -953,6 +985,9 @@ public final class ServerCachingProtos {
if (((bitField0_ & 0x00000008) == 0x00000008)) {
output.writeMessage(4, cacheFactory_);
}
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ output.writeBytes(5, txState_);
+ }
getUnknownFields().writeTo(output);
}
@@ -978,6 +1013,10 @@ public final class ServerCachingProtos {
size += com.google.protobuf.CodedOutputStream
.computeMessageSize(4, cacheFactory_);
}
+ if (((bitField0_ & 0x00000010) == 0x00000010)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(5, txState_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -1021,6 +1060,11 @@ public final class ServerCachingProtos {
result = result && getCacheFactory()
.equals(other.getCacheFactory());
}
+ result = result && (hasTxState() == other.hasTxState());
+ if (hasTxState()) {
+ result = result && getTxState()
+ .equals(other.getTxState());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -1050,6 +1094,10 @@ public final class ServerCachingProtos {
hash = (37 * hash) + CACHEFACTORY_FIELD_NUMBER;
hash = (53 * hash) + getCacheFactory().hashCode();
}
+ if (hasTxState()) {
+ hash = (37 * hash) + TXSTATE_FIELD_NUMBER;
+ hash = (53 * hash) + getTxState().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -1177,6 +1225,8 @@ public final class ServerCachingProtos {
cacheFactoryBuilder_.clear();
}
bitField0_ = (bitField0_ & ~0x00000008);
+ txState_ = com.google.protobuf.ByteString.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00000010);
return this;
}
@@ -1229,6 +1279,10 @@ public final class ServerCachingProtos {
} else {
result.cacheFactory_ = cacheFactoryBuilder_.build();
}
+ if (((from_bitField0_ & 0x00000010) == 0x00000010)) {
+ to_bitField0_ |= 0x00000010;
+ }
+ result.txState_ = txState_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -1257,6 +1311,9 @@ public final class ServerCachingProtos {
if (other.hasCacheFactory()) {
mergeCacheFactory(other.getCacheFactory());
}
+ if (other.hasTxState()) {
+ setTxState(other.getTxState());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -1610,6 +1667,42 @@ public final class ServerCachingProtos {
return cacheFactoryBuilder_;
}
+ // optional bytes txState = 5;
+ private com.google.protobuf.ByteString txState_ = com.google.protobuf.ByteString.EMPTY;
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ public boolean hasTxState() {
+ return ((bitField0_ & 0x00000010) == 0x00000010);
+ }
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ public com.google.protobuf.ByteString getTxState() {
+ return txState_;
+ }
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ public Builder setTxState(com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00000010;
+ txState_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional bytes txState = 5;</code>
+ */
+ public Builder clearTxState() {
+ bitField0_ = (bitField0_ & ~0x00000010);
+ txState_ = getDefaultInstance().getTxState();
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:AddServerCacheRequest)
}
@@ -3383,20 +3476,21 @@ public final class ServerCachingProtos {
"\n\032ServerCachingService.proto\032\030ServerCach" +
"eFactory.proto\"K\n\026ImmutableBytesWritable" +
"\022\021\n\tbyteArray\030\001 \002(\014\022\016\n\006offset\030\002 \002(\005\022\016\n\006l" +
- "ength\030\003 \002(\005\"\220\001\n\025AddServerCacheRequest\022\020\n" +
+ "ength\030\003 \002(\005\"\241\001\n\025AddServerCacheRequest\022\020\n" +
"\010tenantId\030\001 \001(\014\022\017\n\007cacheId\030\002 \002(\014\022)\n\010cach" +
"ePtr\030\003 \002(\0132\027.ImmutableBytesWritable\022)\n\014c" +
- "acheFactory\030\004 \002(\0132\023.ServerCacheFactory\"(" +
- "\n\026AddServerCacheResponse\022\016\n\006return\030\001 \002(\010" +
- "\"=\n\030RemoveServerCacheRequest\022\020\n\010tenantId" +
- "\030\001 \001(\014\022\017\n\007cacheId\030\002 \002(\014\"+\n\031RemoveServerC",
- "acheResponse\022\016\n\006return\030\001 \002(\0102\245\001\n\024ServerC" +
- "achingService\022A\n\016addServerCache\022\026.AddSer" +
- "verCacheRequest\032\027.AddServerCacheResponse" +
- "\022J\n\021removeServerCache\022\031.RemoveServerCach" +
- "eRequest\032\032.RemoveServerCacheResponseBG\n(" +
- "org.apache.phoenix.coprocessor.generated" +
- "B\023ServerCachingProtosH\001\210\001\001\240\001\001"
+ "acheFactory\030\004 \002(\0132\023.ServerCacheFactory\022\017" +
+ "\n\007txState\030\005 \001(\014\"(\n\026AddServerCacheRespons" +
+ "e\022\016\n\006return\030\001 \002(\010\"=\n\030RemoveServerCacheRe" +
+ "quest\022\020\n\010tenantId\030\001 \001(\014\022\017\n\007cacheId\030\002 \002(\014",
+ "\"+\n\031RemoveServerCacheResponse\022\016\n\006return\030" +
+ "\001 \002(\0102\245\001\n\024ServerCachingService\022A\n\016addSer" +
+ "verCache\022\026.AddServerCacheRequest\032\027.AddSe" +
+ "rverCacheResponse\022J\n\021removeServerCache\022\031" +
+ ".RemoveServerCacheRequest\032\032.RemoveServer" +
+ "CacheResponseBG\n(org.apache.phoenix.copr" +
+ "ocessor.generatedB\023ServerCachingProtosH\001" +
+ "\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -3414,7 +3508,7 @@ public final class ServerCachingProtos {
internal_static_AddServerCacheRequest_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_AddServerCacheRequest_descriptor,
- new java.lang.String[] { "TenantId", "CacheId", "CachePtr", "CacheFactory", });
+ new java.lang.String[] { "TenantId", "CacheId", "CachePtr", "CacheFactory", "TxState", });
internal_static_AddServerCacheResponse_descriptor =
getDescriptor().getMessageTypes().get(2);
internal_static_AddServerCacheResponse_fieldAccessorTable = new
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/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 bb76ccb..3613c95 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
@@ -159,6 +159,12 @@ public enum SQLExceptionCode {
AGGREGATE_EXPRESSION_NOT_ALLOWED_IN_INDEX(520, "42897", "Aggregate expression not allowed in an index"),
NON_DETERMINISTIC_EXPRESSION_NOT_ALLOWED_IN_INDEX(521, "42898", "Non-deterministic expression not allowed in an index"),
STATELESS_EXPRESSION_NOT_ALLOWED_IN_INDEX(522, "42899", "Stateless expression not allowed in an index"),
+
+ /**
+ * Transaction exceptions.
+ */
+ TRANSACTION_CONFLICT_EXCEPTION(523, "42900", "Transaction aborted due to conflict with other mutations"),
+ TRANSACTION_EXCEPTION(524, "42901", "Transaction aborted due to error"),
/**
* Union All related errors
@@ -258,7 +264,18 @@ public enum SQLExceptionCode {
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."),
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"),
+
+ DEFAULT_COLUMN_FAMILY_ON_SHARED_TABLE(1069, "43A13", "Default column family not allowed on VIEW or shared INDEX"),
+ ONLY_TABLE_MAY_BE_DECLARED_TRANSACTIONAL(1070, "44A01", "Only tables may be declared as transactional"),
+ TX_MAY_NOT_SWITCH_TO_NON_TX(1071, "44A02", "A transactional table may not be switched to non transactional"),
+ STORE_NULLS_MUST_BE_TRUE_FOR_TRANSACTIONAL(1072, "44A03", "Store nulls must be true when a table is transactional"),
+ CANNOT_START_TRANSACTION_WITH_SCN_SET(1073, "44A04", "Cannot start a transaction on a connection with SCN set"),
+ TX_MAX_VERSIONS_MUST_BE_GREATER_THAN_ONE(1074, "44A05", "A transactional table must define VERSION of greater than one"),
+ CANNOT_SPECIFY_SCN_FOR_TXN_TABLE(1075, "44A06", "Cannot use a connection with SCN set for a transactional table"),
+ NULL_TRANSACTION_CONTEXT(1076, "44A07", "No Tranasction Context available"),
+ TRANSACTION_FAILED(1077, "44A08", "Transaction Failure "),
/** Sequence related */
SEQUENCE_ALREADY_EXIST(1200, "42Z00", "Sequence already exists.", new Factory() {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
index e292dd8..0bdb65a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
@@ -223,23 +223,26 @@ public abstract class BaseQueryPlan implements QueryPlan {
if (table.getType() != PTableType.SYSTEM) {
scan.setConsistency(connection.getConsistency());
}
- // Get the time range of row_timestamp column
- TimeRange rowTimestampRange = context.getScanRanges().getRowTimestampRange();
- // Get the already existing time range on the scan.
- TimeRange scanTimeRange = scan.getTimeRange();
- Long scn = connection.getSCN();
- if (scn == null) {
- scn = context.getCurrentTime();
- }
- try {
- TimeRange timeRangeToUse = ScanUtil.intersectTimeRange(rowTimestampRange, scanTimeRange, scn);
- if (timeRangeToUse == null) {
- return ResultIterator.EMPTY_ITERATOR;
- }
- scan.setTimeRange(timeRangeToUse.getMin(), timeRangeToUse.getMax());
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
+ // TODO fix this in PHOENIX-2415 Support ROW_TIMESTAMP with transactional tables
+ if (!table.isTransactional()) {
+ // Get the time range of row_timestamp column
+ TimeRange rowTimestampRange = context.getScanRanges().getRowTimestampRange();
+ // Get the already existing time range on the scan.
+ TimeRange scanTimeRange = scan.getTimeRange();
+ Long scn = connection.getSCN();
+ if (scn == null) {
+ scn = context.getCurrentTime();
+ }
+ try {
+ TimeRange timeRangeToUse = ScanUtil.intersectTimeRange(rowTimestampRange, scanTimeRange, scn);
+ if (timeRangeToUse == null) {
+ return ResultIterator.EMPTY_ITERATOR;
+ }
+ scan.setTimeRange(timeRangeToUse.getMin(), timeRangeToUse.getMax());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
byte[] tenantIdBytes;
if( table.isMultiTenant() == true ) {
tenantIdBytes = connection.getTenantId() == null ? null :
@@ -326,6 +329,9 @@ public abstract class BaseQueryPlan implements QueryPlan {
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
IndexMaintainer.serialize(dataTable, ptr, indexes, context.getConnection());
scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD, ByteUtil.copyKeyBytesIfNecessary(ptr));
+ if (dataTable.isTransactional()) {
+ scan.setAttribute(BaseScannerRegionObserver.TX_STATE, context.getConnection().getMutationState().encodeTransaction());
+ }
}
private void serializeViewConstantsIntoScan(Scan scan, PTable dataTable) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/execute/DelegateHTable.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/DelegateHTable.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/DelegateHTable.java
new file mode 100644
index 0000000..6b3f9ca
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/DelegateHTable.java
@@ -0,0 +1,300 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.execute;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Append;
+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;
+import org.apache.hadoop.hbase.client.Increment;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Row;
+import org.apache.hadoop.hbase.client.RowMutations;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.coprocessor.Batch.Call;
+import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
+
+import com.google.protobuf.Descriptors.MethodDescriptor;
+import com.google.protobuf.Message;
+import com.google.protobuf.Service;
+import com.google.protobuf.ServiceException;
+
+public class DelegateHTable implements HTableInterface {
+ protected final HTableInterface delegate;
+
+ public DelegateHTable(HTableInterface delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public byte[] getTableName() {
+ return delegate.getTableName();
+ }
+
+ @Override
+ public TableName getName() {
+ return delegate.getName();
+ }
+
+ @Override
+ public Configuration getConfiguration() {
+ return delegate.getConfiguration();
+ }
+
+ @Override
+ public HTableDescriptor getTableDescriptor() throws IOException {
+ return delegate.getTableDescriptor();
+ }
+
+ @Override
+ public boolean exists(Get get) throws IOException {
+ return delegate.exists(get);
+ }
+
+ @Override
+ public Boolean[] exists(List<Get> gets) throws IOException {
+ return delegate.exists(gets);
+ }
+
+ @Override
+ public void batch(List<? extends Row> actions, Object[] results) throws IOException, InterruptedException {
+ delegate.batch(actions, results);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public Object[] batch(List<? extends Row> actions) throws IOException, InterruptedException {
+ return delegate.batch(actions);
+ }
+
+ @Override
+ public <R> void batchCallback(List<? extends Row> actions, Object[] results, Callback<R> callback)
+ throws IOException, InterruptedException {
+ delegate.batchCallback(actions, results, callback);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public <R> Object[] batchCallback(List<? extends Row> actions, Callback<R> callback) throws IOException,
+ InterruptedException {
+ return delegate.batchCallback(actions, callback);
+ }
+
+ @Override
+ public Result get(Get get) throws IOException {
+ return delegate.get(get);
+ }
+
+ @Override
+ public Result[] get(List<Get> gets) throws IOException {
+ return delegate.get(gets);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public Result getRowOrBefore(byte[] row, byte[] family) throws IOException {
+ return delegate.getRowOrBefore(row, family);
+ }
+
+ @Override
+ public ResultScanner getScanner(Scan scan) throws IOException {
+ return delegate.getScanner(scan);
+ }
+
+ @Override
+ public ResultScanner getScanner(byte[] family) throws IOException {
+ return delegate.getScanner(family);
+ }
+
+ @Override
+ public ResultScanner getScanner(byte[] family, byte[] qualifier) throws IOException {
+ return delegate.getScanner(family, qualifier);
+ }
+
+ @Override
+ public void put(Put put) throws IOException {
+ delegate.put(put);
+ }
+
+ @Override
+ public void put(List<Put> puts) throws IOException {
+ delegate.put(puts);
+ }
+
+ @Override
+ public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier, byte[] value, Put put) throws IOException {
+ return delegate.checkAndPut(row, family, qualifier, value, put);
+ }
+
+ @Override
+ public void delete(Delete delete) throws IOException {
+ delegate.delete(delete);
+ }
+
+ @Override
+ public void delete(List<Delete> deletes) throws IOException {
+ delegate.delete(deletes);
+ }
+
+ @Override
+ public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier, byte[] value, Delete delete)
+ throws IOException {
+ return delegate.checkAndDelete(row, family, qualifier, value, delete);
+ }
+
+ @Override
+ public void mutateRow(RowMutations rm) throws IOException {
+ delegate.mutateRow(rm);
+ }
+
+ @Override
+ public Result append(Append append) throws IOException {
+ return delegate.append(append);
+ }
+
+ @Override
+ public Result increment(Increment increment) throws IOException {
+ return delegate.increment(increment);
+ }
+
+ @Override
+ public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount) throws IOException {
+ return delegate.incrementColumnValue(row, family, qualifier, amount);
+ }
+
+ @Override
+ public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount, Durability durability)
+ throws IOException {
+ return delegate.incrementColumnValue(row, family, qualifier, amount, durability);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public long incrementColumnValue(byte[] row, byte[] family, byte[] qualifier, long amount, boolean writeToWAL)
+ throws IOException {
+ return delegate.incrementColumnValue(row, family, qualifier, amount, writeToWAL);
+ }
+
+ @Override
+ public boolean isAutoFlush() {
+ return delegate.isAutoFlush();
+ }
+
+ @Override
+ public void flushCommits() throws IOException {
+ delegate.flushCommits();
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+
+ @Override
+ public CoprocessorRpcChannel coprocessorService(byte[] row) {
+ return delegate.coprocessorService(row);
+ }
+
+ @Override
+ public <T extends Service, R> Map<byte[], R> coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
+ Call<T, R> callable) throws ServiceException, Throwable {
+ return delegate.coprocessorService(service, startKey, endKey, callable);
+ }
+
+ @Override
+ public <T extends Service, R> void coprocessorService(Class<T> service, byte[] startKey, byte[] endKey,
+ Call<T, R> callable, Callback<R> callback) throws ServiceException, Throwable {
+ delegate.coprocessorService(service, startKey, endKey, callable, callback);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public void setAutoFlush(boolean autoFlush) {
+ delegate.setAutoFlush(autoFlush);
+ }
+
+ @Override
+ public void setAutoFlush(boolean autoFlush, boolean clearBufferOnFail) {
+ delegate.setAutoFlush(autoFlush, clearBufferOnFail);
+ }
+
+ @Override
+ public void setAutoFlushTo(boolean autoFlush) {
+ delegate.setAutoFlushTo(autoFlush);
+ }
+
+ @Override
+ public long getWriteBufferSize() {
+ return delegate.getWriteBufferSize();
+ }
+
+ @Override
+ public void setWriteBufferSize(long writeBufferSize) throws IOException {
+ delegate.setWriteBufferSize(writeBufferSize);
+ }
+
+ @Override
+ public <R extends Message> Map<byte[], R> batchCoprocessorService(MethodDescriptor methodDescriptor,
+ Message request, byte[] startKey, byte[] endKey, R responsePrototype) throws ServiceException, Throwable {
+ return delegate.batchCoprocessorService(methodDescriptor, request, startKey, endKey, responsePrototype);
+ }
+
+ @Override
+ public <R extends Message> void batchCoprocessorService(MethodDescriptor methodDescriptor, Message request,
+ byte[] startKey, byte[] endKey, R responsePrototype, Callback<R> callback) throws ServiceException,
+ Throwable {
+ delegate.batchCoprocessorService(methodDescriptor, request, startKey, endKey, responsePrototype, callback);
+ }
+
+ @Override
+ public boolean checkAndMutate(byte[] row, byte[] family, byte[] qualifier, CompareOp compareOp, byte[] value,
+ RowMutations mutation) throws IOException {
+ return delegate.checkAndMutate(row, family, qualifier, compareOp, value, mutation);
+ }
+
+ @Override
+ public boolean[] existsAll(List<Get> gets) throws IOException {
+ return delegate.existsAll(gets);
+ }
+
+ @Override
+ public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier,
+ CompareOp compareOp, byte[] value, Put put) throws IOException {
+ return delegate.checkAndPut(row, family, qualifier, value, put);
+ }
+
+ @Override
+ public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier,
+ CompareOp compareOp, byte[] value, Delete delete)
+ throws IOException {
+ return delegate.checkAndDelete(row, family, qualifier, compareOp, value, delete);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc9929b5/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
index 72920b2..cf89380 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
@@ -23,6 +23,7 @@ import static org.apache.phoenix.util.LogUtil.addCustomAnnotations;
import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -66,6 +67,7 @@ import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PArrayDataType;
import org.apache.phoenix.schema.types.PBoolean;
@@ -75,6 +77,7 @@ import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.SQLCloseables;
import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
public class HashJoinPlan extends DelegateQueryPlan {
private static final Log LOG = LogFactory.getLog(HashJoinPlan.class);
@@ -83,6 +86,7 @@ public class HashJoinPlan extends DelegateQueryPlan {
private final HashJoinInfo joinInfo;
private final SubPlan[] subPlans;
private final boolean recompileWhereClause;
+ private final Set<TableRef> tableRefs;
private final int maxServerCacheTimeToLive;
private List<SQLCloseable> dependencies;
private HashCacheClient hashClient;
@@ -114,11 +118,21 @@ public class HashJoinPlan extends DelegateQueryPlan {
this.joinInfo = joinInfo;
this.subPlans = subPlans;
this.recompileWhereClause = recompileWhereClause;
+ this.tableRefs = Sets.newHashSetWithExpectedSize(subPlans.length + plan.getSourceRefs().size());
+ this.tableRefs.addAll(plan.getSourceRefs());
+ for (SubPlan subPlan : subPlans) {
+ tableRefs.addAll(subPlan.getInnerPlan().getSourceRefs());
+ }
this.maxServerCacheTimeToLive = plan.getContext().getConnection().getQueryServices().getProps().getInt(
QueryServices.MAX_SERVER_CACHE_TIME_TO_LIVE_MS_ATTRIB, QueryServicesOptions.DEFAULT_MAX_SERVER_CACHE_TIME_TO_LIVE_MS);
}
@Override
+ public Set<TableRef> getSourceRefs() {
+ return tableRefs;
+ }
+
+ @Override
public ResultIterator iterator() throws SQLException {
return iterator(DefaultParallelScanGrouper.getInstance());
}
@@ -251,6 +265,7 @@ public class HashJoinPlan extends DelegateQueryPlan {
public void postProcess(Object result, HashJoinPlan parent) throws SQLException;
public List<String> getPreSteps(HashJoinPlan parent) throws SQLException;
public List<String> getPostSteps(HashJoinPlan parent) throws SQLException;
+ public QueryPlan getInnerPlan();
}
public static class WhereClauseSubPlan implements SubPlan {
@@ -321,6 +336,11 @@ public class HashJoinPlan extends DelegateQueryPlan {
public List<String> getPostSteps(HashJoinPlan parent) throws SQLException {
return Collections.<String>emptyList();
}
+
+ @Override
+ public QueryPlan getInnerPlan() {
+ return plan;
+ }
}
public static class HashSubPlan implements SubPlan {
@@ -411,7 +431,12 @@ public class HashJoinPlan extends DelegateQueryPlan {
+ " IN (" + keyRangeRhsExpression.toString() + ")";
return Collections.<String> singletonList(step);
}
-
+
+
+ @Override
+ public QueryPlan getInnerPlan() {
+ return plan;
+ }
}
}