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/12/01 05:18:29 UTC
[04/19] 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/4a70c701/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
new file mode 100644
index 0000000..75ad63d
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
@@ -0,0 +1,505 @@
+/*
+ * 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.index;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellScanner;
+import org.apache.hadoop.hbase.CoprocessorEnvironment;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValueUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.cloudera.htrace.Span;
+import org.cloudera.htrace.Trace;
+import org.cloudera.htrace.TraceScope;
+import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.hbase.index.MultiMutation;
+import org.apache.phoenix.hbase.index.ValueGetter;
+import org.apache.phoenix.hbase.index.covered.IndexUpdate;
+import org.apache.phoenix.hbase.index.covered.TableState;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+import org.apache.phoenix.hbase.index.covered.update.ColumnTracker;
+import org.apache.phoenix.hbase.index.covered.update.IndexedColumnGroup;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.hbase.index.write.IndexWriter;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.trace.TracingUtils;
+import org.apache.phoenix.trace.util.NullSpan;
+import org.apache.phoenix.util.ScanUtil;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.ServerUtil;
+
+import co.cask.tephra.Transaction;
+import co.cask.tephra.Transaction.VisibilityLevel;
+import co.cask.tephra.TxConstants;
+import co.cask.tephra.hbase98.TransactionAwareHTable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.primitives.Longs;
+
+/**
+ * Do all the work of managing index updates for a transactional table from a single coprocessor. Since the transaction
+ * manager essentially time orders writes through conflict detection, the logic to maintain a secondary index is quite a
+ * bit simpler than the non transactional case. For example, there's no need to muck with the WAL, as failure scenarios
+ * are handled by aborting the transaction.
+ */
+public class PhoenixTransactionalIndexer extends BaseRegionObserver {
+
+ private static final Log LOG = LogFactory.getLog(PhoenixTransactionalIndexer.class);
+
+ private PhoenixIndexCodec codec;
+ private IndexWriter writer;
+ private boolean stopped;
+
+ @Override
+ public void start(CoprocessorEnvironment e) throws IOException {
+ final RegionCoprocessorEnvironment env = (RegionCoprocessorEnvironment)e;
+ String serverName = env.getRegionServerServices().getServerName().getServerName();
+ codec = new PhoenixIndexCodec();
+ codec.initialize(env);
+
+ // setup the actual index writer
+ this.writer = new IndexWriter(env, serverName + "-tx-index-writer");
+ }
+
+ @Override
+ public void stop(CoprocessorEnvironment e) throws IOException {
+ if (this.stopped) { return; }
+ this.stopped = true;
+ String msg = "TxIndexer is being stopped";
+ this.writer.stop(msg);
+ }
+
+ private static Iterator<Mutation> getMutationIterator(final MiniBatchOperationInProgress<Mutation> miniBatchOp) {
+ return new Iterator<Mutation>() {
+ private int i = 0;
+
+ @Override
+ public boolean hasNext() {
+ return i < miniBatchOp.size();
+ }
+
+ @Override
+ public Mutation next() {
+ return miniBatchOp.getOperation(i++);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ };
+ }
+ @Override
+ public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
+
+ Mutation m = miniBatchOp.getOperation(0);
+ if (!codec.isEnabled(m)) {
+ super.preBatchMutate(c, miniBatchOp);
+ return;
+ }
+
+ Map<String,byte[]> updateAttributes = m.getAttributesMap();
+ PhoenixIndexMetaData indexMetaData = new PhoenixIndexMetaData(c.getEnvironment(),updateAttributes);
+ byte[] txRollbackAttribute = m.getAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY);
+ Collection<Pair<Mutation, byte[]>> indexUpdates = null;
+ // get the current span, or just use a null-span to avoid a bunch of if statements
+ try (TraceScope scope = Trace.startSpan("Starting to build index updates")) {
+ Span current = scope.getSpan();
+ if (current == null) {
+ current = NullSpan.INSTANCE;
+ }
+
+ // get the index updates for all elements in this batch
+ indexUpdates = getIndexUpdates(c.getEnvironment(), indexMetaData, getMutationIterator(miniBatchOp), txRollbackAttribute);
+
+ current.addTimelineAnnotation("Built index updates, doing preStep");
+ TracingUtils.addAnnotation(current, "index update count", indexUpdates.size());
+
+ // no index updates, so we are done
+ if (!indexUpdates.isEmpty()) {
+ this.writer.write(indexUpdates);
+ }
+ } catch (Throwable t) {
+ String msg = "Failed to update index with entries:" + indexUpdates;
+ LOG.error(msg, t);
+ ServerUtil.throwIOException(msg, t);
+ }
+ }
+
+ private Collection<Pair<Mutation, byte[]>> getIndexUpdates(RegionCoprocessorEnvironment env, PhoenixIndexMetaData indexMetaData, Iterator<Mutation> mutationIterator, byte[] txRollbackAttribute) throws IOException {
+ ResultScanner currentScanner = null;
+ TransactionAwareHTable txTable = null;
+ // Collect up all mutations in batch
+ Map<ImmutableBytesPtr, MultiMutation> mutations =
+ new HashMap<ImmutableBytesPtr, MultiMutation>();
+ while(mutationIterator.hasNext()) {
+ Mutation m = mutationIterator.next();
+ // add the mutation to the batch set
+ ImmutableBytesPtr row = new ImmutableBytesPtr(m.getRow());
+ MultiMutation stored = mutations.get(row);
+ // we haven't seen this row before, so add it
+ if (stored == null) {
+ stored = new MultiMutation(row);
+ mutations.put(row, stored);
+ }
+ stored.addAll(m);
+ }
+
+ // Collect the set of mutable ColumnReferences so that we can first
+ // run a scan to get the current state. We'll need this to delete
+ // the existing index rows.
+ Transaction tx = indexMetaData.getTransaction();
+ assert(tx != null);
+ List<IndexMaintainer> indexMaintainers = indexMetaData.getIndexMaintainers();
+ Set<ColumnReference> mutableColumns = Sets.newHashSetWithExpectedSize(indexMaintainers.size() * 10);
+ for (IndexMaintainer indexMaintainer : indexMaintainers) {
+ // For transactional tables, we use an index maintainer
+ // to aid in rollback if there's a KeyValue column in the index. The alternative would be
+ // to hold on to all uncommitted index row keys (even ones already sent to HBase) on the
+ // client side.
+ mutableColumns.addAll(indexMaintainer.getAllColumns());
+ }
+
+ boolean isRollback = txRollbackAttribute!=null;
+ Collection<Pair<Mutation, byte[]>> indexUpdates = new ArrayList<Pair<Mutation, byte[]>>(mutations.size() * 2 * indexMaintainers.size());
+ try {
+ if (!mutableColumns.isEmpty()) {
+ List<KeyRange> keys = Lists.newArrayListWithExpectedSize(mutations.size());
+ for (ImmutableBytesPtr ptr : mutations.keySet()) {
+ keys.add(PVarbinary.INSTANCE.getKeyRange(ptr.copyBytesIfNecessary()));
+ }
+ Scan scan = new Scan();
+ // Project all mutable columns
+ for (ColumnReference ref : mutableColumns) {
+ scan.addColumn(ref.getFamily(), ref.getQualifier());
+ }
+ // Project empty key value column
+ scan.addColumn(indexMaintainers.get(0).getDataEmptyKeyValueCF(), QueryConstants.EMPTY_COLUMN_BYTES);
+ ScanRanges scanRanges = ScanRanges.create(SchemaUtil.VAR_BINARY_SCHEMA, Collections.singletonList(keys), ScanUtil.SINGLE_COLUMN_SLOT_SPAN, KeyRange.EVERYTHING_RANGE, null, true, -1);
+ scanRanges.initializeScan(scan);
+ scan.setFilter(scanRanges.getSkipScanFilter());
+ TableName tableName = env.getRegion().getRegionInfo().getTable();
+ HTableInterface htable = env.getTable(tableName);
+ txTable = new TransactionAwareHTable(htable);
+ txTable.startTx(tx);
+ // For rollback, we need to see all versions, including
+ // the last committed version as there may be multiple
+ // checkpointed versions.
+ if (isRollback) {
+ tx.setVisibility(VisibilityLevel.SNAPSHOT_ALL);
+ }
+ currentScanner = txTable.getScanner(scan);
+ }
+ if (isRollback) {
+ processRollback(env, indexMetaData, txRollbackAttribute, currentScanner, mutations, tx, mutableColumns, indexUpdates);
+ } else {
+ processMutation(env, indexMetaData, txRollbackAttribute, currentScanner, mutations, tx, mutableColumns, indexUpdates);
+ }
+ } finally {
+ if (txTable != null) txTable.close();
+ }
+
+ return indexUpdates;
+ }
+
+ private void processMutation(RegionCoprocessorEnvironment env,
+ PhoenixIndexMetaData indexMetaData, byte[] txRollbackAttribute,
+ ResultScanner scanner,
+ Map<ImmutableBytesPtr, MultiMutation> mutations, Transaction tx,
+ Set<ColumnReference> mutableColumns,
+ Collection<Pair<Mutation, byte[]>> indexUpdates) throws IOException {
+ if (scanner != null) {
+ Result result;
+ ColumnReference emptyColRef = new ColumnReference(indexMetaData.getIndexMaintainers().get(0).getDataEmptyKeyValueCF(), QueryConstants.EMPTY_COLUMN_BYTES);
+ // Process existing data table rows by removing the old index row and adding the new index row
+ while ((result = scanner.next()) != null) {
+ Mutation m = mutations.remove(new ImmutableBytesPtr(result.getRow()));
+ TxTableState state = new TxTableState(env, mutableColumns, indexMetaData.getAttributes(), tx.getWritePointer(), m, emptyColRef, result);
+ generateDeletes(indexMetaData, indexUpdates, txRollbackAttribute, state);
+ generatePuts(indexMetaData, indexUpdates, state);
+ }
+ }
+ // Process new data table by adding new index rows
+ for (Mutation m : mutations.values()) {
+ TxTableState state = new TxTableState(env, mutableColumns, indexMetaData.getAttributes(), tx.getWritePointer(), m);
+ generatePuts(indexMetaData, indexUpdates, state);
+ }
+ }
+
+ private void processRollback(RegionCoprocessorEnvironment env,
+ PhoenixIndexMetaData indexMetaData, byte[] txRollbackAttribute,
+ ResultScanner scanner,
+ Map<ImmutableBytesPtr, MultiMutation> mutations, Transaction tx,
+ Set<ColumnReference> mutableColumns,
+ Collection<Pair<Mutation, byte[]>> indexUpdates) throws IOException {
+ if (scanner != null) {
+ Result result;
+ // Loop through last committed row state plus all new rows associated with current transaction
+ // to generate point delete markers for all index rows that were added. We don't have Tephra
+ // manage index rows in change sets because we don't want to be hit with the additional
+ // memory hit and do not need to do conflict detection on index rows.
+ ColumnReference emptyColRef = new ColumnReference(indexMetaData.getIndexMaintainers().get(0).getDataEmptyKeyValueCF(), QueryConstants.EMPTY_COLUMN_BYTES);
+ while ((result = scanner.next()) != null) {
+ Mutation m = mutations.remove(new ImmutableBytesPtr(result.getRow()));
+ // Sort by timestamp, type, cf, cq so we can process in time batches from oldest to newest
+ // (as if we're "replaying" them in time order).
+ List<Cell> cells = result.listCells();
+ Collections.sort(cells, new Comparator<Cell>() {
+
+ @Override
+ public int compare(Cell o1, Cell o2) {
+ int c = Longs.compare(o1.getTimestamp(), o2.getTimestamp());
+ if (c != 0) return c;
+ c = o1.getTypeByte() - o2.getTypeByte();
+ if (c != 0) return c;
+ c = Bytes.compareTo(o1.getFamilyArray(), o1.getFamilyOffset(), o1.getFamilyLength(), o1.getFamilyArray(), o1.getFamilyOffset(), o1.getFamilyLength());
+ if (c != 0) return c;
+ return Bytes.compareTo(o1.getQualifierArray(), o1.getQualifierOffset(), o1.getQualifierLength(), o1.getQualifierArray(), o1.getQualifierOffset(), o1.getQualifierLength());
+ }
+
+ });
+ int i = 0;
+ int nCells = cells.size();
+ Result oldResult = null, newResult;
+ long readPtr = tx.getReadPointer();
+ do {
+ boolean hasPuts = false;
+ LinkedList<Cell> singleTimeCells = Lists.newLinkedList();
+ long writePtr;
+ do {
+ Cell cell = cells.get(i);
+ hasPuts |= cell.getTypeByte() == KeyValue.Type.Put.getCode();
+ writePtr = cell.getTimestamp();
+ do {
+ // Add at the beginning of the list to match the expected HBase
+ // newest to oldest sort order (which TxTableState relies on
+ // with the Result.getLatestColumnValue() calls).
+ singleTimeCells.addFirst(cell);
+ } while (++i < nCells && cells.get(i).getTimestamp() == writePtr);
+ } while (i < nCells && cells.get(i).getTimestamp() <= readPtr);
+
+ // Generate point delete markers for the prior row deletion of the old index value.
+ // The write timestamp is the next timestamp, not the current timestamp,
+ // as the earliest cells are the current values for the row (and we don't
+ // want to delete the current row).
+ if (oldResult != null) {
+ TxTableState state = new TxTableState(env, mutableColumns, indexMetaData.getAttributes(), writePtr, m, emptyColRef, oldResult);
+ generateDeletes(indexMetaData, indexUpdates, txRollbackAttribute, state);
+ }
+ // Generate point delete markers for the new index value.
+ // If our time batch doesn't have Puts (i.e. we have only Deletes), then do not
+ // generate deletes. We would have generated the delete above based on the state
+ // of the previous row. The delete markers do not give us the state we need to
+ // delete.
+ if (hasPuts) {
+ newResult = Result.create(singleTimeCells);
+ // First row may represent the current state which we don't want to delete
+ if (writePtr > readPtr) {
+ TxTableState state = new TxTableState(env, mutableColumns, indexMetaData.getAttributes(), writePtr, m, emptyColRef, newResult);
+ generateDeletes(indexMetaData, indexUpdates, txRollbackAttribute, state);
+ }
+ oldResult = newResult;
+ } else {
+ oldResult = null;
+ }
+ } while (i < nCells);
+ }
+ }
+ }
+
+ private void generateDeletes(PhoenixIndexMetaData indexMetaData,
+ Collection<Pair<Mutation, byte[]>> indexUpdates,
+ byte[] attribValue, TxTableState state) throws IOException {
+ Iterable<IndexUpdate> deletes = codec.getIndexDeletes(state, indexMetaData);
+ for (IndexUpdate delete : deletes) {
+ if (delete.isValid()) {
+ delete.getUpdate().setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, attribValue);
+ indexUpdates.add(new Pair<Mutation, byte[]>(delete.getUpdate(),delete.getTableName()));
+ }
+ }
+ }
+
+ private boolean generatePuts(
+ PhoenixIndexMetaData indexMetaData,
+ Collection<Pair<Mutation, byte[]>> indexUpdates,
+ TxTableState state)
+ throws IOException {
+ state.applyMutation();
+ Iterable<IndexUpdate> puts = codec.getIndexUpserts(state, indexMetaData);
+ boolean validPut = false;
+ for (IndexUpdate put : puts) {
+ if (put.isValid()) {
+ indexUpdates.add(new Pair<Mutation, byte[]>(put.getUpdate(),put.getTableName()));
+ validPut = true;
+ }
+ }
+ return validPut;
+ }
+
+
+ private static class TxTableState implements TableState {
+ private final Mutation mutation;
+ private final long currentTimestamp;
+ private final RegionCoprocessorEnvironment env;
+ private final Map<String, byte[]> attributes;
+ private final List<KeyValue> pendingUpdates;
+ private final Set<ColumnReference> indexedColumns;
+ private final Map<ColumnReference, ImmutableBytesWritable> valueMap;
+
+ private TxTableState(RegionCoprocessorEnvironment env, Set<ColumnReference> indexedColumns, Map<String, byte[]> attributes, long currentTimestamp, Mutation mutation) {
+ this.env = env;
+ this.currentTimestamp = currentTimestamp;
+ this.indexedColumns = indexedColumns;
+ this.attributes = attributes;
+ this.mutation = mutation;
+ int estimatedSize = indexedColumns.size();
+ this.valueMap = Maps.newHashMapWithExpectedSize(estimatedSize);
+ this.pendingUpdates = Lists.newArrayListWithExpectedSize(estimatedSize);
+ try {
+ CellScanner scanner = mutation.cellScanner();
+ while (scanner.advance()) {
+ Cell cell = scanner.current();
+ pendingUpdates.add(KeyValueUtil.ensureKeyValue(cell));
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e); // Impossible
+ }
+ }
+
+ public TxTableState(RegionCoprocessorEnvironment env, Set<ColumnReference> indexedColumns, Map<String, byte[]> attributes, long currentTimestamp, Mutation m, ColumnReference emptyColRef, Result r) {
+ this(env, indexedColumns, attributes, currentTimestamp, m);
+
+ for (ColumnReference ref : indexedColumns) {
+ Cell cell = r.getColumnLatestCell(ref.getFamily(), ref.getQualifier());
+ if (cell != null) {
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ ptr.set(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ valueMap.put(ref, ptr);
+ }
+ }
+ }
+
+ @Override
+ public RegionCoprocessorEnvironment getEnvironment() {
+ return env;
+ }
+
+ @Override
+ public long getCurrentTimestamp() {
+ return currentTimestamp;
+ }
+
+ @Override
+ public Map<String, byte[]> getUpdateAttributes() {
+ return attributes;
+ }
+
+ @Override
+ public byte[] getCurrentRowKey() {
+ return mutation.getRow();
+ }
+
+ @Override
+ public List<? extends IndexedColumnGroup> getIndexColumnHints() {
+ return Collections.emptyList();
+ }
+
+ private void applyMutation() {
+ for (Cell cell : pendingUpdates) {
+ if (cell.getTypeByte() == KeyValue.Type.Delete.getCode() || cell.getTypeByte() == KeyValue.Type.DeleteColumn.getCode()) {
+ ColumnReference ref = new ColumnReference(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(), cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
+ valueMap.remove(ref);
+ } else if (cell.getTypeByte() == KeyValue.Type.DeleteFamily.getCode() || cell.getTypeByte() == KeyValue.Type.DeleteFamilyVersion.getCode()) {
+ for (ColumnReference ref : indexedColumns) {
+ if (ref.matchesFamily(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength())) {
+ valueMap.remove(ref);
+ }
+ }
+ } else if (cell.getTypeByte() == KeyValue.Type.Put.getCode()){
+ ColumnReference ref = new ColumnReference(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength(), cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
+ if (indexedColumns.contains(ref)) {
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ ptr.set(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
+ valueMap.put(ref, ptr);
+ }
+ } else {
+ throw new IllegalStateException("Unexpected mutation type for " + cell);
+ }
+ }
+ }
+
+ @Override
+ public Collection<KeyValue> getPendingUpdate() {
+ return pendingUpdates;
+ }
+
+ @Override
+ public Pair<ValueGetter, IndexUpdate> getIndexUpdateState(Collection<? extends ColumnReference> indexedColumns)
+ throws IOException {
+ // TODO: creating these objects over and over again is wasteful
+ ColumnTracker tracker = new ColumnTracker(indexedColumns);
+ ValueGetter getter = new ValueGetter() {
+
+ @Override
+ public ImmutableBytesWritable getLatestValue(ColumnReference ref) throws IOException {
+ return valueMap.get(ref);
+ }
+
+ @Override
+ public byte[] getRowKey() {
+ return mutation.getRow();
+ }
+
+ };
+ Pair<ValueGetter, IndexUpdate> pair = new Pair<ValueGetter, IndexUpdate>(getter, new IndexUpdate(tracker));
+ return pair;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index 57e84f0..c18233e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -55,6 +55,7 @@ import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.filter.ColumnProjectionFilter;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.parse.FilterableStatement;
@@ -104,6 +105,7 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
private final byte[] physicalTableName;
private final QueryPlan plan;
protected final String scanId;
+ protected final MutationState mutationState;
private final ParallelScanGrouper scanGrouper;
// TODO: too much nesting here - breakup into new classes.
private final List<List<List<Pair<Scan,Future<PeekingResultIterator>>>>> allFutures;
@@ -202,6 +204,9 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
this.plan = plan;
this.scanGrouper = scanGrouper;
StatementContext context = plan.getContext();
+ // Clone MutationState as the one on the connection will change if auto commit is on
+ // yet we need the original one with the original transaction from TableResultIterator.
+ this.mutationState = new MutationState(context.getConnection().getMutationState());
TableRef tableRef = plan.getTableRef();
PTable table = tableRef.getTable();
physicalTableName = table.getPhysicalName().getBytes();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
index f272e55..ef8c4a6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.TableRef;
@@ -51,38 +52,44 @@ public class ChunkedResultIterator implements PeekingResultIterator {
private ImmutableBytesWritable lastKey = new ImmutableBytesWritable();
private final StatementContext context;
private final TableRef tableRef;
- private Scan scan;
private final long chunkSize;
+ private final MutationState mutationState;
+ private Scan scan;
private PeekingResultIterator resultIterator;
public static class ChunkedResultIteratorFactory implements ParallelIteratorFactory {
private final ParallelIteratorFactory delegateFactory;
private final TableRef tableRef;
+ private final MutationState mutationState;
public ChunkedResultIteratorFactory(ParallelIteratorFactory
- delegateFactory, TableRef tableRef) {
+ delegateFactory, MutationState mutationState, TableRef tableRef) {
this.delegateFactory = delegateFactory;
this.tableRef = tableRef;
+ // Clone MutationState, as the one on the connection may change if auto commit is on
+ // while we need a handle to the original one (for it's transaction state).
+ this.mutationState = new MutationState(mutationState);
}
@Override
public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan, String tableName) throws SQLException {
if (logger.isDebugEnabled()) logger.debug(LogUtil.addCustomAnnotations("ChunkedResultIteratorFactory.newIterator over " + tableRef.getTable().getName().getString() + " with " + scan, ScanUtil.getCustomAnnotations(scan)));
- return new ChunkedResultIterator(delegateFactory, context, tableRef, scan,
- context.getConnection().getQueryServices().getProps().getLong(
- QueryServices.SCAN_RESULT_CHUNK_SIZE,
- QueryServicesOptions.DEFAULT_SCAN_RESULT_CHUNK_SIZE), scanner);
+ return new ChunkedResultIterator(delegateFactory, mutationState, context, tableRef, scan,
+ mutationState.getConnection().getQueryServices().getProps().getLong(
+ QueryServices.SCAN_RESULT_CHUNK_SIZE,
+ QueryServicesOptions.DEFAULT_SCAN_RESULT_CHUNK_SIZE), scanner);
}
}
- private ChunkedResultIterator(ParallelIteratorFactory delegateIteratorFactory,
- StatementContext context, TableRef tableRef, Scan scan, long chunkSize, ResultIterator scanner) throws SQLException {
+ private ChunkedResultIterator(ParallelIteratorFactory delegateIteratorFactory, MutationState mutationState,
+ StatementContext context, TableRef tableRef, Scan scan, long chunkSize, ResultIterator scanner) throws SQLException {
this.delegateIteratorFactory = delegateIteratorFactory;
this.context = context;
this.tableRef = tableRef;
this.scan = scan;
this.chunkSize = chunkSize;
+ this.mutationState = mutationState;
// Instantiate single chunk iterator and the delegate iterator in constructor
// to get parallel scans kicked off in separate threads. If we delay this,
// we'll get serialized behavior (see PHOENIX-
@@ -120,7 +127,7 @@ public class ChunkedResultIterator implements PeekingResultIterator {
if (logger.isDebugEnabled()) logger.debug(LogUtil.addCustomAnnotations("Get next chunked result iterator over " + tableRef.getTable().getName().getString() + " with " + scan, ScanUtil.getCustomAnnotations(scan)));
String tableName = tableRef.getTable().getPhysicalName().getString();
ResultIterator singleChunkResultIterator = new SingleChunkResultIterator(
- new TableResultIterator(context, tableRef, scan, context.getReadMetricsQueue().allotMetric(SCAN_BYTES, tableName)), chunkSize);
+ new TableResultIterator(mutationState, tableRef, scan, context.getReadMetricsQueue().allotMetric(SCAN_BYTES, tableName)), chunkSize);
resultIterator = delegateIteratorFactory.newIterator(context, singleChunkResultIterator, scan, tableName);
}
return resultIterator;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index 87f8335..265831c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -102,7 +102,7 @@ public class ParallelIterators extends BaseResultIterators {
@Override
public PeekingResultIterator call() throws Exception {
long startTime = System.currentTimeMillis();
- ResultIterator scanner = new TableResultIterator(context, tableRef, scan, scanMetrics);
+ ResultIterator scanner = new TableResultIterator(mutationState, tableRef, scan, scanMetrics);
if (logger.isDebugEnabled()) {
logger.debug(LogUtil.addCustomAnnotations("Id: " + scanId + ", Time: " + (System.currentTimeMillis() - startTime) + "ms, Scan: " + scan, ScanUtil.getCustomAnnotations(scan)));
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
index fa18c83..e0936b2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
@@ -80,7 +80,7 @@ public class SerialIterators extends BaseResultIterators {
public PeekingResultIterator call() throws Exception {
List<PeekingResultIterator> concatIterators = Lists.newArrayListWithExpectedSize(scans.size());
for (final Scan scan : scans) {
- ResultIterator scanner = new TableResultIterator(context, tableRef, scan, context.getReadMetricsQueue().allotMetric(SCAN_BYTES, tableName), ScannerCreation.DELAYED);
+ ResultIterator scanner = new TableResultIterator(mutationState, tableRef, scan, context.getReadMetricsQueue().allotMetric(SCAN_BYTES, tableName), ScannerCreation.DELAYED);
concatIterators.add(iteratorFactory.newIterator(context, scanner, scan, tableName));
}
PeekingResultIterator concatIterator = ConcatResultIterator.newIterator(concatIterators);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
index 6f040d1..7d07a4a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
@@ -22,8 +22,11 @@ import java.sql.SQLException;
import java.util.List;
import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
-import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.monitoring.CombinableMetric;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.tuple.Tuple;
@@ -39,7 +42,7 @@ import org.apache.phoenix.util.ServerUtil;
*
* @since 0.1
*/
-public class TableResultIterator extends ExplainTable implements ResultIterator {
+public class TableResultIterator implements ResultIterator {
public enum ScannerCreation {IMMEDIATE, DELAYED};
private final Scan scan;
@@ -47,8 +50,8 @@ public class TableResultIterator extends ExplainTable implements ResultIterator
private volatile ResultIterator delegate;
private final CombinableMetric scanMetrics;
- public TableResultIterator(StatementContext context, TableRef tableRef, CombinableMetric scanMetrics) throws SQLException {
- this(context, tableRef, context.getScan(), scanMetrics);
+ public TableResultIterator(MutationState mutationState, Scan scan, TableRef tableRef, CombinableMetric scanMetrics) throws SQLException {
+ this(mutationState, tableRef, scan, scanMetrics);
}
/*
@@ -75,15 +78,15 @@ public class TableResultIterator extends ExplainTable implements ResultIterator
return delegate;
}
- public TableResultIterator(StatementContext context, TableRef tableRef, Scan scan, CombinableMetric scanMetrics) throws SQLException {
- this(context, tableRef, scan, scanMetrics, ScannerCreation.IMMEDIATE);
+ public TableResultIterator(MutationState mutationState, TableRef tableRef, Scan scan, CombinableMetric scanMetrics) throws SQLException {
+ this(mutationState, tableRef, scan, scanMetrics, ScannerCreation.IMMEDIATE);
}
- public TableResultIterator(StatementContext context, TableRef tableRef, Scan scan, CombinableMetric scanMetrics, ScannerCreation creationMode) throws SQLException {
- super(context, tableRef);
+ public TableResultIterator(MutationState mutationState, TableRef tableRef, Scan scan, CombinableMetric scanMetrics, ScannerCreation creationMode) throws SQLException {
this.scan = scan;
this.scanMetrics = scanMetrics;
- htable = context.getConnection().getQueryServices().getTable(tableRef.getTable().getPhysicalName().getBytes());
+ PTable table = tableRef.getTable();
+ htable = mutationState.getHTable(table);
if (creationMode == ScannerCreation.IMMEDIATE) {
getDelegate(false);
}
@@ -109,8 +112,11 @@ public class TableResultIterator extends ExplainTable implements ResultIterator
@Override
public void explain(List<String> planSteps) {
- StringBuilder buf = new StringBuilder();
- explain(buf.toString(),planSteps);
+ try {
+ getDelegate(false).explain(planSteps);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index 2d06487..68312c6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -76,6 +76,7 @@ import org.apache.phoenix.schema.PMetaData.Pruner;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTableKey;
+import org.apache.phoenix.schema.PTableRef;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.types.PArrayDataType;
@@ -155,14 +156,18 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
}
public PhoenixConnection(PhoenixConnection connection, boolean isDescRowKeyOrderUpgrade) throws SQLException {
- this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.getMetaDataCache(), isDescRowKeyOrderUpgrade);
+ this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.metaData, connection.getMutationState(), isDescRowKeyOrderUpgrade);
this.isAutoCommit = connection.isAutoCommit;
this.sampler = connection.sampler;
this.statementExecutionCounter = connection.statementExecutionCounter;
}
public PhoenixConnection(PhoenixConnection connection) throws SQLException {
- this(connection, connection.isDescVarLengthRowKeyUpgrade);
+ this(connection, connection.isDescVarLengthRowKeyUpgrade());
+ }
+
+ public PhoenixConnection(PhoenixConnection connection, MutationState mutationState) throws SQLException {
+ this(connection.getQueryServices(), connection.getURL(), connection.getClientInfo(), connection.getMetaDataCache(), mutationState, connection.isDescVarLengthRowKeyUpgrade());
}
public PhoenixConnection(PhoenixConnection connection, long scn) throws SQLException {
@@ -170,17 +175,21 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
}
public PhoenixConnection(ConnectionQueryServices services, PhoenixConnection connection, long scn) throws SQLException {
- this(services, connection.getURL(), newPropsWithSCN(scn,connection.getClientInfo()), connection.getMetaDataCache(), connection.isDescVarLengthRowKeyUpgrade());
+ this(services, connection.getURL(), newPropsWithSCN(scn,connection.getClientInfo()), connection.metaData, connection.getMutationState(), connection.isDescVarLengthRowKeyUpgrade());
this.isAutoCommit = connection.isAutoCommit;
this.sampler = connection.sampler;
this.statementExecutionCounter = connection.statementExecutionCounter;
}
public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData) throws SQLException {
- this(services, url, info, metaData, false);
+ this(services, url, info, metaData, null, false);
}
-
- public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData, boolean isDescVarLengthRowKeyUpgrade) throws SQLException {
+
+ public PhoenixConnection(PhoenixConnection connection, ConnectionQueryServices services, Properties info) throws SQLException {
+ this(services, connection.url, info, connection.metaData, null, connection.isDescVarLengthRowKeyUpgrade());
+ }
+
+ public PhoenixConnection(ConnectionQueryServices services, String url, Properties info, PMetaData metaData, MutationState mutationState, boolean isDescVarLengthRowKeyUpgrade) throws SQLException {
this.url = url;
this.isDescVarLengthRowKeyUpgrade = isDescVarLengthRowKeyUpgrade;
// Copy so client cannot change
@@ -249,7 +258,7 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
}
};
this.isRequestLevelMetricsEnabled = JDBCUtil.isCollectingRequestLevelMetricsEnabled(url, info, this.services.getProps());
- this.mutationState = newMutationState(maxSize);
+ this.mutationState = mutationState == null ? newMutationState(maxSize) : new MutationState(mutationState);
this.metaData = metaData.pruneTables(pruner);
this.metaData = metaData.pruneFunctions(pruner);
this.services.addConnection(this);
@@ -402,12 +411,16 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
return mutateBatchSize;
}
+ public PMetaData getMetaDataCache() {
+ return metaData;
+ }
+
public PTable getTable(PTableKey key) throws TableNotFoundException {
- return metaData.getTable(key);
+ return metaData.getTableRef(key).getTable();
}
- public PMetaData getMetaDataCache() {
- return metaData;
+ public PTableRef getTableRef(PTableKey key) throws TableNotFoundException {
+ return metaData.getTableRef(key);
}
protected MutationState newMutationState(int maxSize) {
@@ -444,7 +457,9 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
// from modifying this list.
this.statements = Lists.newArrayList();
try {
- mutationState.rollback(this);
+ mutationState.rollback();
+ } catch (SQLException e) {
+ // ignore any exceptions while rolling back
} finally {
try {
SQLCloseables.closeAll(statements);
@@ -683,7 +698,13 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
@Override
public void rollback() throws SQLException {
- mutationState.rollback(this);
+ CallRunner.run(new CallRunner.CallableThrowable<Void, SQLException>() {
+ @Override
+ public Void call() throws SQLException {
+ mutationState.rollback();
+ return null;
+ }
+ }, Tracing.withTracing(this, "rolling back"));
statementExecutionCounter = 0;
}
@@ -792,16 +813,20 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
// TODO Auto-generated method stub
return 0;
}
-
+
@Override
- public PMetaData addTable(PTable table) throws SQLException {
- // TODO: since a connection is only used by one thread at a time,
- // we could modify this metadata in place since it's not shared.
- if (scn == null || scn > table.getTimeStamp()) {
- metaData = metaData.addTable(table);
- }
+ public PMetaData addTable(PTable table, long resolvedTime) throws SQLException {
+ metaData = metaData.addTable(table, resolvedTime);
//Cascade through to connectionQueryServices too
- getQueryServices().addTable(table);
+ getQueryServices().addTable(table, resolvedTime);
+ return metaData;
+ }
+
+ @Override
+ public PMetaData updateResolvedTimestamp(PTable table, long resolvedTime) throws SQLException {
+ metaData = metaData.updateResolvedTimestamp(table, resolvedTime);
+ //Cascade through to connectionQueryServices too
+ getQueryServices().updateResolvedTimestamp(table, resolvedTime);
return metaData;
}
@@ -818,11 +843,11 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
}
@Override
- public PMetaData addColumn(PName tenantId, String tableName, List<PColumn> columns, long tableTimeStamp, long tableSeqNum, boolean isImmutableRows, boolean isWalDisabled, boolean isMultitenant, boolean storeNulls)
+ public PMetaData addColumn(PName tenantId, String tableName, List<PColumn> columns, long tableTimeStamp, long tableSeqNum, boolean isImmutableRows, boolean isWalDisabled, boolean isMultitenant, boolean storeNulls, boolean isTransactional, long resolvedTime)
throws SQLException {
- metaData = metaData.addColumn(tenantId, tableName, columns, tableTimeStamp, tableSeqNum, isImmutableRows, isWalDisabled, isMultitenant, storeNulls);
+ metaData = metaData.addColumn(tenantId, tableName, columns, tableTimeStamp, tableSeqNum, isImmutableRows, isWalDisabled, isMultitenant, storeNulls, isTransactional, resolvedTime);
//Cascade through to connectionQueryServices too
- getQueryServices().addColumn(tenantId, tableName, columns, tableTimeStamp, tableSeqNum, isImmutableRows, isWalDisabled, isMultitenant, storeNulls);
+ getQueryServices().addColumn(tenantId, tableName, columns, tableTimeStamp, tableSeqNum, isImmutableRows, isWalDisabled, isMultitenant, storeNulls, isTransactional, resolvedTime);
return metaData;
}
@@ -844,10 +869,10 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
@Override
public PMetaData removeColumn(PName tenantId, String tableName, List<PColumn> columnsToRemove, long tableTimeStamp,
- long tableSeqNum) throws SQLException {
- metaData = metaData.removeColumn(tenantId, tableName, columnsToRemove, tableTimeStamp, tableSeqNum);
+ long tableSeqNum, long resolvedTime) throws SQLException {
+ metaData = metaData.removeColumn(tenantId, tableName, columnsToRemove, tableTimeStamp, tableSeqNum, resolvedTime);
//Cascade through to connectionQueryServices too
- getQueryServices().removeColumn(tenantId, tableName, columnsToRemove, tableTimeStamp, tableSeqNum);
+ getQueryServices().removeColumn(tenantId, tableName, columnsToRemove, tableTimeStamp, tableSeqNum, resolvedTime);
return metaData;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/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 3f6958c..d6d1230 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
@@ -301,6 +301,9 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
private static final String TENANT_POS_SHIFT = "TENANT_POS_SHIFT";
private static final byte[] TENANT_POS_SHIFT_BYTES = Bytes.toBytes(TENANT_POS_SHIFT);
+ public static final String TRANSACTIONAL = "TRANSACTIONAL";
+ public static final byte[] TRANSACTIONAL_BYTES = Bytes.toBytes(TRANSACTIONAL);
+
private final PhoenixConnection connection;
private final ResultSet emptyResultSet;
public static final int MAX_LOCAL_SI_VERSION_DISALLOW = VersionUtil.encodeVersion("0", "98", "8");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java
index 1a785a6..bb0b1e2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java
@@ -162,7 +162,7 @@ public final class PhoenixDriver extends PhoenixEmbeddedDriver {
ConnectionQueryServices connectionQueryServices = connectionQueryServicesMap.get(normalizedConnInfo);
if (connectionQueryServices == null) {
if (normalizedConnInfo.isConnectionless()) {
- connectionQueryServices = new ConnectionlessQueryServicesImpl(services, normalizedConnInfo);
+ connectionQueryServices = new ConnectionlessQueryServicesImpl(services, normalizedConnInfo, info);
} else {
connectionQueryServices = new ConnectionQueryServicesImpl(services, normalizedConnInfo, info);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
index 3cfaacc..8ee9dbd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
@@ -187,6 +187,10 @@ public abstract class PhoenixEmbeddedDriver implements Driver, org.apache.phoeni
.setMessage(url).build().buildException();
}
+ public String getZookeeperConnectionString() {
+ return getZookeeperQuorum() + ":" + getPort();
+ }
+
/**
* Detect url with quorum:1,quorum:2 as HBase does not handle different port numbers
* for different quorum hostnames.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index b2b058f..76836d5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -34,6 +34,7 @@ import java.text.Format;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -260,6 +261,12 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
final long startTime = System.currentTimeMillis();
try {
QueryPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
+ // Send mutations to hbase, so they are visible to subsequent reads.
+ // Use original plan for data table so that data and immutable indexes will be sent
+ // TODO: for joins, we need to iterate through all tables, but we need the original table,
+ // not the projected table, so plan.getContext().getResolver().getTables() won't work.
+ Iterator<TableRef> tableRefs = plan.getSourceRefs().iterator();
+ connection.getMutationState().sendUncommitted(tableRefs);
plan = connection.getQueryServices().getOptimizer().optimize(PhoenixStatement.this, plan);
// this will create its own trace internally, so we don't wrap this
// whole thing in tracing
@@ -319,6 +326,12 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
try {
MutationState state = connection.getMutationState();
MutationPlan plan = stmt.compilePlan(PhoenixStatement.this, Sequence.ValueOp.VALIDATE_SEQUENCE);
+ if (plan.getTargetRef() != null && plan.getTargetRef().getTable() != null && plan.getTargetRef().getTable().isTransactional()) {
+ state.startTransaction();
+ }
+ Iterator<TableRef> tableRefs = plan.getSourceRefs().iterator();
+ state.sendUncommitted(tableRefs);
+ state.checkpointIfNeccessary(plan);
MutationState lastState = plan.execute();
state.join(lastState);
if (connection.getAutoCommit()) {
@@ -1556,4 +1569,4 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java b/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
index f13b28e..e8b3389 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheClient.java
@@ -39,6 +39,7 @@ import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.ServerUtil;
import org.apache.phoenix.util.TrustedByteArrayOutputStream;
import org.apache.phoenix.util.TupleUtil;
@@ -80,7 +81,7 @@ public class HashCacheClient {
*/
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
serialize(ptr, iterator, estimatedSize, onExpressions, singleValueOnly, keyRangeRhsExpression, keyRangeRhsValues);
- return serverCache.addServerCache(keyRanges, ptr, new HashCacheFactory(), cacheUsingTableRef);
+ return serverCache.addServerCache(keyRanges, ptr, ByteUtil.EMPTY_BYTE_ARRAY, new HashCacheFactory(), cacheUsingTableRef);
}
private void serialize(ImmutableBytesWritable ptr, ResultIterator iterator, long estimatedSize, List<Expression> onExpressions, boolean singleValueOnly, Expression keyRangeRhsExpression, List<Expression> keyRangeRhsValues) throws SQLException {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheFactory.java
index f1f6e22..908a117 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/join/HashCacheFactory.java
@@ -69,7 +69,7 @@ public class HashCacheFactory implements ServerCacheFactory {
}
@Override
- public Closeable newCache(ImmutableBytesWritable cachePtr, MemoryChunk chunk) throws SQLException {
+ public Closeable newCache(ImmutableBytesWritable cachePtr, byte[] txState, MemoryChunk chunk) throws SQLException {
try {
// This reads the uncompressed length from the front of the compressed input
int uncompressedLen = Snappy.getUncompressedLength(cachePtr.get(), cachePtr.getOffset());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
index b500a25..d2f91b3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
@@ -108,7 +108,7 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
ReadMetricQueue readMetrics = ctx.getReadMetricsQueue();
String tableName = queryPlan.getTableRef().getTable().getPhysicalName().getString();
for (Scan scan : scans) {
- final TableResultIterator tableResultIterator = new TableResultIterator(queryPlan.getContext(),
+ final TableResultIterator tableResultIterator = new TableResultIterator(queryPlan.getContext().getConnection().getMutationState(),
queryPlan.getTableRef(), scan, readMetrics.allotMetric(SCAN_BYTES, tableName));
PeekingResultIterator peekingResultIterator = LookAheadResultIterator.wrap(tableResultIterator);
iterators.add(peekingResultIterator);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
index 32b66f1..3bc3808 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/PhoenixIndexImportDirectMapper.java
@@ -138,7 +138,7 @@ public class PhoenixIndexImportDirectMapper extends
private void writeBatch(MutationState mutationState, Context context) throws IOException,
SQLException, InterruptedException {
- final Iterator<Pair<byte[], List<Mutation>>> iterator = mutationState.toMutations(true);
+ final Iterator<Pair<byte[], List<Mutation>>> iterator = mutationState.toMutations(true, null);
while (iterator.hasNext()) {
Pair<byte[], List<Mutation>> mutationPair = iterator.next();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java b/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
index bc318d2..c5ff3c6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
@@ -95,17 +95,11 @@ public class QueryOptimizer {
return plans.get(0);
}
- public List<QueryPlan> getBestPlan(PhoenixStatement statement, SelectStatement select, ColumnResolver resolver, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory) throws SQLException {
- return getApplicablePlans(statement, select, resolver, targetColumns, parallelIteratorFactory, true);
+ public List<QueryPlan> getBestPlan(QueryPlan dataPlan, PhoenixStatement statement, SelectStatement select, ColumnResolver resolver, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory) throws SQLException {
+ return getApplicablePlans(dataPlan, statement, targetColumns, parallelIteratorFactory, true);
}
- public List<QueryPlan> getApplicablePlans(PhoenixStatement statement, SelectStatement select, ColumnResolver resolver, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory) throws SQLException {
- return getApplicablePlans(statement, select, resolver, targetColumns, parallelIteratorFactory, false);
- }
-
- private List<QueryPlan> getApplicablePlans(PhoenixStatement statement, SelectStatement select, ColumnResolver resolver, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory, boolean stopAtBestPlan) throws SQLException {
- QueryCompiler compiler = new QueryCompiler(statement, select, resolver, targetColumns, parallelIteratorFactory, new SequenceManager(statement));
- QueryPlan dataPlan = compiler.compile();
+ public List<QueryPlan> getApplicablePlans(QueryPlan dataPlan, PhoenixStatement statement, SelectStatement select, ColumnResolver resolver, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory) throws SQLException {
return getApplicablePlans(dataPlan, statement, targetColumns, parallelIteratorFactory, false);
}
@@ -221,7 +215,8 @@ public class QueryOptimizer {
private static QueryPlan addPlan(PhoenixStatement statement, SelectStatement select, PTable index, List<? extends PDatum> targetColumns, ParallelIteratorFactory parallelIteratorFactory, QueryPlan dataPlan, boolean isHinted) throws SQLException {
int nColumns = dataPlan.getProjector().getColumnCount();
- String alias = '"' + dataPlan.getTableRef().getTableAlias() + '"'; // double quote in case it's case sensitive
+ String tableAlias = dataPlan.getTableRef().getTableAlias();
+ String alias = tableAlias==null ? null : '"' + tableAlias + '"'; // double quote in case it's case sensitive
String schemaName = index.getParentSchemaName().getString();
schemaName = schemaName.length() == 0 ? null : '"' + schemaName + '"';
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4a70c701/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
index 3b53309..08c417f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
@@ -44,6 +44,8 @@ import org.apache.phoenix.schema.SequenceAllocation;
import org.apache.phoenix.schema.SequenceKey;
import org.apache.phoenix.schema.stats.PTableStats;
+import co.cask.tephra.TransactionSystemClient;
+
public interface ConnectionQueryServices extends QueryServices, MetaDataMutated {
public static final int INITIAL_META_DATA_TABLE_CAPACITY = 100;
@@ -88,7 +90,7 @@ public interface ConnectionQueryServices extends QueryServices, MetaDataMutated
void clearTableRegionCache(byte[] tableName) throws SQLException;
- boolean hasInvalidIndexConfiguration();
+ boolean isMutableIndexWALCodecInstalled();
long createSequence(String tenantId, String schemaName, String sequenceName, long startWith, long incrementBy, long cacheSize, long minValue, long maxValue, boolean cycle, long timestamp) throws SQLException;
long dropSequence(String tenantId, String schemaName, String sequenceName, long timestamp) throws SQLException;
@@ -116,4 +118,6 @@ public interface ConnectionQueryServices extends QueryServices, MetaDataMutated
public void clearCache() throws SQLException;
public int getSequenceSaltBuckets();
+
+ TransactionSystemClient getTransactionSystemClient();
}
\ No newline at end of file