PHOENIX-5494 Batched, mutable Index updates are unnecessarily run one-by-one
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java
index a909cf2..5571512 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/BaseIndexBuilder.java
@@ -133,4 +133,8 @@
public ReplayWrite getReplayWrite(Mutation m) {
return null;
}
+
+ public RegionCoprocessorEnvironment getEnv() {
+ return this.env;
+ }
}
\ No newline at end of file
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java
index 165f0c9..43cafa0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuildManager.java
@@ -21,7 +21,6 @@
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
-
import com.google.common.collect.ListMultimap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
@@ -37,6 +36,7 @@
import org.apache.phoenix.hbase.index.covered.IndexMetaData;
import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.hbase.index.covered.data.CachedLocalTable;
import org.apache.phoenix.index.PhoenixIndexMetaData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -49,6 +49,7 @@
private static final Logger LOGGER = LoggerFactory.getLogger(IndexBuildManager.class);
private final IndexBuilder delegate;
private boolean stopped;
+ private RegionCoprocessorEnvironment regionCoprocessorEnvironment;
/**
* @param env environment in which <tt>this</tt> is running. Used to setup the
@@ -59,6 +60,7 @@
// Prevent deadlock by using single thread for all reads so that we know
// we can get the ReentrantRWLock. See PHOENIX-2671 for more details.
this.delegate = getIndexBuilder(env);
+ this.regionCoprocessorEnvironment = env;
}
private static IndexBuilder getIndexBuilder(RegionCoprocessorEnvironment e) throws IOException {
@@ -88,10 +90,14 @@
IndexMetaData indexMetaData) throws Throwable {
// notify the delegate that we have started processing a batch
this.delegate.batchStarted(miniBatchOp, indexMetaData);
-
+ CachedLocalTable cachedLocalTable =
+ CachedLocalTable.build(
+ mutations,
+ (PhoenixIndexMetaData)indexMetaData,
+ this.regionCoprocessorEnvironment.getRegion());
// Avoid the Object overhead of the executor when it's not actually parallelizing anything.
for (Mutation m : mutations) {
- Collection<Pair<Mutation, byte[]>> updates = delegate.getIndexUpdate(m, indexMetaData);
+ Collection<Pair<Mutation, byte[]>> updates = delegate.getIndexUpdate(m, indexMetaData, cachedLocalTable);
for (Pair<Mutation, byte[]> update : updates) {
indexUpdates.put(new HTableInterfaceReference(new ImmutableBytesPtr(update.getSecond())), new Pair<>(update.getFirst(), m.getRow()));
}
@@ -105,10 +111,15 @@
final IndexMetaData indexMetaData = this.delegate.getIndexMetaData(miniBatchOp);
this.delegate.batchStarted(miniBatchOp, indexMetaData);
+ CachedLocalTable cachedLocalTable =
+ CachedLocalTable.build(
+ mutations,
+ (PhoenixIndexMetaData)indexMetaData,
+ this.regionCoprocessorEnvironment.getRegion());
// Avoid the Object overhead of the executor when it's not actually parallelizing anything.
ArrayList<Pair<Mutation, byte[]>> results = new ArrayList<>(mutations.size());
for (Mutation m : mutations) {
- Collection<Pair<Mutation, byte[]>> updates = delegate.getIndexUpdate(m, indexMetaData);
+ Collection<Pair<Mutation, byte[]>> updates = delegate.getIndexUpdate(m, indexMetaData, cachedLocalTable);
if (PhoenixIndexMetaData.isIndexRebuild(m.getAttributesMap())) {
for (Pair<Mutation, byte[]> update : updates) {
update.getFirst().setAttribute(BaseScannerRegionObserver.REPLAY_WRITES,
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java
index f561e79..7bce22e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/builder/IndexBuilder.java
@@ -34,6 +34,7 @@
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver.ReplayWrite;
import org.apache.phoenix.hbase.index.Indexer;
import org.apache.phoenix.hbase.index.covered.IndexMetaData;
+import org.apache.phoenix.hbase.index.covered.data.LocalHBaseState;
/**
* Interface to build updates ({@link Mutation}s) to the index tables, based on the primary table
@@ -72,7 +73,7 @@
* @return a Map of the mutations to make -> target index table name
* @throws IOException on failure
*/
- public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Mutation mutation, IndexMetaData context) throws IOException;
+ public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Mutation mutation, IndexMetaData context, LocalHBaseState localHBaseState) throws IOException;
/**
* Build an index update to cleanup the index when we remove {@link KeyValue}s via the normal flush or compaction
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
index 54d7f87..bf592ba 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/LocalTableState.java
@@ -161,7 +161,7 @@
// needing to lookup the prior row values.
if (requiresPriorRowState) {
// add the current state of the row. Uses listCells() to avoid a new array creation.
- this.addUpdateCells(this.table.getCurrentRowState(update, toCover, ignoreNewerMutations).listCells(), false);
+ this.addUpdateCells(this.table.getCurrentRowState(update, toCover, ignoreNewerMutations), false);
}
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
index 645f2c4..63d2751 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
@@ -20,7 +20,6 @@
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.hbase.index.builder.BaseIndexBuilder;
import org.apache.phoenix.hbase.index.covered.data.LocalHBaseState;
-import org.apache.phoenix.hbase.index.covered.data.LocalTable;
import org.apache.phoenix.hbase.index.covered.update.ColumnTracker;
import org.apache.phoenix.hbase.index.covered.update.IndexUpdateManager;
import org.slf4j.Logger;
@@ -38,18 +37,15 @@
public class NonTxIndexBuilder extends BaseIndexBuilder {
private static final Logger LOGGER = LoggerFactory.getLogger(NonTxIndexBuilder.class);
- protected LocalHBaseState localTable;
-
@Override
public void setup(RegionCoprocessorEnvironment env) throws IOException {
super.setup(env);
- this.localTable = new LocalTable(env);
}
@Override
- public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Mutation mutation, IndexMetaData indexMetaData) throws IOException {
+ public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Mutation mutation, IndexMetaData indexMetaData, LocalHBaseState localHBaseState) throws IOException {
// create a state manager, so we can manage each batch
- LocalTableState state = new LocalTableState(localTable, mutation);
+ LocalTableState state = new LocalTableState(localHBaseState, mutation);
// build the index updates for each group
IndexUpdateManager manager = new IndexUpdateManager(indexMetaData);
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/CachedLocalTable.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/CachedLocalTable.java
new file mode 100644
index 0000000..2fd91f7
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/CachedLocalTable.java
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.hbase.index.covered.data;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.filter.SkipScanFilter;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.index.PhoenixIndexMetaData;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.schema.types.PVarbinary;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Sets;
+
+import java.util.HashMap;
+
+public class CachedLocalTable implements LocalHBaseState {
+
+ private final HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells;
+ private final Region region;
+
+ private CachedLocalTable(HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells, Region region) {
+ this.rowKeyPtrToCells = rowKeyPtrToCells;
+ this.region = region;
+ }
+
+ @Override
+ public List<Cell> getCurrentRowState(
+ Mutation mutation,
+ Collection<? extends ColumnReference> columnReferences,
+ boolean ignoreNewerMutations) throws IOException {
+
+ if(ignoreNewerMutations) {
+ return doScan(mutation, columnReferences);
+ }
+
+ byte[] rowKey = mutation.getRow();
+ return this.rowKeyPtrToCells.get(new ImmutableBytesPtr(rowKey));
+ }
+
+ private List<Cell> doScan(Mutation mutation, Collection<? extends ColumnReference> columnReferences) throws IOException {
+ byte[] rowKey = mutation.getRow();
+ // need to use a scan here so we can get raw state, which Get doesn't provide.
+ Scan scan = IndexManagementUtil.newLocalStateScan(Collections.singletonList(columnReferences));
+ scan.setStartRow(rowKey);
+ scan.setStopRow(rowKey);
+
+ // Provides a means of client indicating that newer cells should not be considered,
+ // enabling mutations to be replayed to partially rebuild the index when a write fails.
+ // When replaying mutations we want the oldest timestamp (as anything newer we be replayed)
+ //long ts = getOldestTimestamp(m.getFamilyCellMap().values());
+ long ts = getMutationTimestampWhenAllCellTimestampIsSame(mutation);
+ scan.setTimeRange(0,ts);
+
+ try (RegionScanner regionScanner = region.getScanner(scan)) {
+ List<Cell> cells = new ArrayList<Cell>(1);
+ boolean more = regionScanner.next(cells);
+ assert !more : "Got more than one result when scanning"
+ + " a single row in the primary table!";
+
+ return cells;
+ }
+ }
+
+ @VisibleForTesting
+ public static CachedLocalTable build(HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells) {
+ return new CachedLocalTable(rowKeyPtrToCells, null);
+ }
+
+ public static CachedLocalTable build(
+ Collection<? extends Mutation> dataTableMutationsWithSameRowKeyAndTimestamp,
+ final PhoenixIndexMetaData indexMetaData,
+ Region region) throws IOException {
+ if(indexMetaData.getReplayWrite() != null)
+ {
+ return new CachedLocalTable(new HashMap<ImmutableBytesPtr,List<Cell>>(), region);
+ }
+ return preScanAllRequiredRows(dataTableMutationsWithSameRowKeyAndTimestamp, indexMetaData, region);
+ }
+
+ /**
+ * Pre-scan all the required rows before we building the indexes for the dataTableMutationsWithSameRowKeyAndTimestamp
+ * parameter.
+ * Note: When we calling this method, for single mutation in the dataTableMutationsWithSameRowKeyAndTimestamp
+ * parameter, all cells in the mutation have the same rowKey and timestamp.
+ * @param dataTableMutationsWithSameRowKeyAndTimestamp
+ * @param indexMetaData
+ * @param region
+ * @throws IOException
+ */
+ public static CachedLocalTable preScanAllRequiredRows(
+ Collection<? extends Mutation> dataTableMutationsWithSameRowKeyAndTimestamp,
+ PhoenixIndexMetaData indexMetaData,
+ Region region) throws IOException {
+ List<IndexMaintainer> indexTableMaintainers = indexMetaData.getIndexMaintainers();
+ Set<KeyRange> keys = new HashSet<KeyRange>(dataTableMutationsWithSameRowKeyAndTimestamp.size());
+ for (Mutation mutation : dataTableMutationsWithSameRowKeyAndTimestamp) {
+ keys.add(PVarbinary.INSTANCE.getKeyRange(mutation.getRow()));
+ }
+
+ Set<ColumnReference> getterColumnReferences = Sets.newHashSet();
+ for (IndexMaintainer indexTableMaintainer : indexTableMaintainers) {
+ getterColumnReferences.addAll(
+ indexTableMaintainer.getAllColumns());
+ }
+
+ getterColumnReferences.add(new ColumnReference(
+ indexTableMaintainers.get(0).getDataEmptyKeyValueCF(),
+ indexTableMaintainers.get(0).getEmptyKeyValueQualifier()));
+
+ Scan scan = IndexManagementUtil.newLocalStateScan(
+ Collections.singletonList(getterColumnReferences));
+ ScanRanges scanRanges = ScanRanges.createPointLookup(new ArrayList<KeyRange>(keys));
+ scanRanges.initializeScan(scan);
+ SkipScanFilter skipScanFilter = scanRanges.getSkipScanFilter();
+
+ if(indexMetaData.getReplayWrite() != null) {
+ /**
+ * Because of previous {@link IndexManagementUtil#flattenMutationsByTimestamp}(which is called
+ * in {@link IndexRegionObserver#groupMutations} or {@link Indexer#preBatchMutateWithExceptions}),
+ * for single mutation in the dataTableMutationsWithSameRowKeyAndTimestamp, all cells in the mutation
+ * have the same rowKey and timestamp.
+ */
+ long timestamp = getMaxTimestamp(dataTableMutationsWithSameRowKeyAndTimestamp);
+ scan.setTimeRange(0, timestamp);
+ scan.setFilter(new SkipScanFilter(skipScanFilter, true));
+ } else {
+ assert scan.isRaw();
+ scan.setMaxVersions(1);
+ scan.setFilter(skipScanFilter);
+ }
+
+ HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells =
+ new HashMap<ImmutableBytesPtr, List<Cell>>();
+ try (RegionScanner scanner = region.getScanner(scan)) {
+ boolean more = true;
+ while(more) {
+ List<Cell> cells = new ArrayList<Cell>();
+ more = scanner.next(cells);
+ if (cells.isEmpty()) {
+ continue;
+ }
+ Cell cell = cells.get(0);
+ byte[] rowKey = CellUtil.cloneRow(cell);
+ rowKeyPtrToCells.put(new ImmutableBytesPtr(rowKey), cells);
+ }
+ }
+
+ return new CachedLocalTable(rowKeyPtrToCells, region);
+ }
+
+ private static long getMaxTimestamp(Collection<? extends Mutation> dataTableMutationsWithSameRowKeyAndTimestamp) {
+ long maxTimestamp = Long.MIN_VALUE;
+ for(Mutation mutation : dataTableMutationsWithSameRowKeyAndTimestamp) {
+ /**
+ * all the cells in this mutation have the same timestamp.
+ */
+ long timestamp = getMutationTimestampWhenAllCellTimestampIsSame(mutation);
+ if(timestamp > maxTimestamp) {
+ maxTimestamp = timestamp;
+ }
+ }
+ return maxTimestamp;
+ }
+
+ private static long getMutationTimestampWhenAllCellTimestampIsSame(Mutation mutation) {
+ return mutation.getFamilyCellMap().values().iterator().next().get(0).getTimestamp();
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalHBaseState.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalHBaseState.java
index 5b06910..aae0763 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalHBaseState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalHBaseState.java
@@ -19,6 +19,7 @@
import java.io.IOException;
import java.util.Collection;
+import java.util.List;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.client.Mutation;
@@ -43,7 +44,7 @@
* {@link Result} with no stored {@link Cell}s.
* @throws IOException if there is an issue reading the row
*/
- public Result getCurrentRowState(Mutation m, Collection<? extends ColumnReference> toCover, boolean ignoreNewerMutations)
+ public List<Cell> getCurrentRowState(Mutation m, Collection<? extends ColumnReference> toCover, boolean ignoreNewerMutations)
throws IOException;
}
\ No newline at end of file
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalTable.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalTable.java
deleted file mode 100644
index 402620f..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/data/LocalTable.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.phoenix.hbase.index.covered.data;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.client.Mutation;
-import org.apache.hadoop.hbase.client.Result;
-import org.apache.hadoop.hbase.client.Scan;
-import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.regionserver.Region;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
-import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
-import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
-
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Ordering;
-import com.google.common.primitives.Longs;
-
-/**
- * Wrapper around a lazily instantiated, local HTable.
- * <p>
- * Previously, we had used various row and batch caches. However, this ends up being very
- * complicated when attempting manage updating and invalidating the cache with no real gain as any
- * row accessed multiple times will likely be in HBase's block cache, invalidating any extra caching
- * we are doing here. In the end, its simpler and about as efficient to just get the current state
- * of the row from HBase and let HBase manage caching the row from disk on its own.
- */
-public class LocalTable implements LocalHBaseState {
-
- private RegionCoprocessorEnvironment env;
-
- public LocalTable(RegionCoprocessorEnvironment env) {
- this.env = env;
- }
-
- @Override
- public Result getCurrentRowState(Mutation m, Collection<? extends ColumnReference> columns, boolean ignoreNewerMutations)
- throws IOException {
- byte[] row = m.getRow();
- // need to use a scan here so we can get raw state, which Get doesn't provide.
- Scan s = IndexManagementUtil.newLocalStateScan(Collections.singletonList(columns));
- s.setStartRow(row);
- s.setStopRow(row);
- if (ignoreNewerMutations) {
- // Provides a means of client indicating that newer cells should not be considered,
- // enabling mutations to be replayed to partially rebuild the index when a write fails.
- // When replaying mutations we want the oldest timestamp (as anything newer we be replayed)
- long ts = getOldestTimestamp(m.getFamilyCellMap().values());
- s.setTimeRange(0,ts);
- }
- Region region = this.env.getRegion();
- try (RegionScanner scanner = region.getScanner(s)) {
- List<Cell> kvs = new ArrayList<Cell>(1);
- boolean more = scanner.next(kvs);
- assert !more : "Got more than one result when scanning"
- + " a single row in the primary table!";
-
- Result r = Result.create(kvs);
- return r;
- }
- }
-
- // Returns the smallest timestamp in the given cell lists.
- // It is assumed that the lists have cells ordered from largest to smallest timestamp
- protected long getOldestTimestamp(Collection<List<Cell>> cellLists) {
- Ordering<List<Cell>> cellListOrdering = new Ordering<List<Cell>>() {
- @Override
- public int compare(List<Cell> left, List<Cell> right) {
- // compare the last element of each list, since that is the smallest in that list
- return Longs.compare(Iterables.getLast(left).getTimestamp(),
- Iterables.getLast(right).getTimestamp());
- }
- };
- List<Cell> minList = cellListOrdering.min(cellLists);
- return Iterables.getLast(minList).getTimestamp();
- }
-}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java
index f07caf7..5a2b467 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java
@@ -20,36 +20,32 @@
import static org.junit.Assert.assertEquals;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValue.Type;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.Region;
-import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver.ReplayWrite;
-import org.apache.phoenix.hbase.index.covered.data.LocalHBaseState;
-import org.apache.phoenix.hbase.index.covered.data.LocalTable;
+import org.apache.phoenix.hbase.index.covered.data.CachedLocalTable;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.scanner.Scanner;
import org.apache.phoenix.hbase.index.scanner.ScannerBuilder.CoveredDeleteScanner;
-import org.apache.phoenix.util.PhoenixKeyValueUtil;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.util.ScanUtil;
import org.junit.Test;
import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-/**
- *
- */
+
public class LocalTableStateTest {
private static final byte[] row = Bytes.toBytes("row");
@@ -88,23 +84,16 @@
Region region = Mockito.mock(Region.class);
Mockito.when(env.getRegion()).thenReturn(region);
- RegionScanner scanner = Mockito.mock(RegionScanner.class);
- Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenReturn(scanner);
final byte[] stored = Bytes.toBytes("stored-value");
- Mockito.when(scanner.next(Mockito.any(List.class))).thenAnswer(new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocation) throws Throwable {
- List<KeyValue> list = (List<KeyValue>) invocation.getArguments()[0];
- KeyValue kv = new KeyValue(row, fam, qual, ts, Type.Put, stored);
- kv.setSequenceId(0);
- list.add(kv);
- return false;
- }
- });
- LocalHBaseState state = new LocalTable(env);
- LocalTableState table = new LocalTableState(state, m);
+ KeyValue kv = new KeyValue(row, fam, qual, ts, Type.Put, stored);
+ kv.setSequenceId(0);
+ HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells =
+ new HashMap<ImmutableBytesPtr, List<Cell>>();
+ rowKeyPtrToCells.put(new ImmutableBytesPtr(row), Collections.singletonList((Cell)kv));
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(rowKeyPtrToCells);
+ LocalTableState table = new LocalTableState(cachedLocalTable, m);
//add the kvs from the mutation
table.addPendingUpdates(m.get(fam, qual));
@@ -122,48 +111,6 @@
super(msg);
}
}
-
- @Test(expected = ScannerCreatedException.class)
- public void testScannerForMutableRows() throws Exception {
- IndexMetaData indexMetaData = new IndexMetaData() {
-
- @Override
- public ReplayWrite getReplayWrite() {
- return null;
- }
-
- @Override
- public boolean requiresPriorRowState(Mutation m) {
- return true;
- }
-
- @Override
- public int getClientVersion() {
- return ScanUtil.UNKNOWN_CLIENT_VERSION;
- }
-
- };
- Put m = new Put(row);
- m.addColumn(fam, qual, ts, val);
- // setup mocks
- Configuration conf = new Configuration(false);
- RegionCoprocessorEnvironment env = Mockito.mock(RegionCoprocessorEnvironment.class);
- Mockito.when(env.getConfiguration()).thenReturn(conf);
-
- Region region = Mockito.mock(Region.class);
- Mockito.when(env.getRegion()).thenReturn(region);
- Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenThrow(new ScannerCreatedException("Should not open scanner when data is immutable"));
-
- LocalHBaseState state = new LocalTable(env);
- LocalTableState table = new LocalTableState(state, m);
- //add the kvs from the mutation
- table.addPendingUpdates(m.get(fam, qual));
-
- // setup the lookup
- ColumnReference col = new ColumnReference(fam, qual);
- table.setCurrentTimestamp(ts);
- table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
- }
@Test
public void testNoScannerForImmutableRows() throws Exception {
@@ -194,10 +141,9 @@
Region region = Mockito.mock(Region.class);
Mockito.when(env.getRegion()).thenReturn(region);
- Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenThrow(new ScannerCreatedException("Should not open scanner when data is immutable"));
- LocalHBaseState state = new LocalTable(env);
- LocalTableState table = new LocalTableState(state, m);
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(null);
+ LocalTableState table = new LocalTableState(cachedLocalTable, m);
//add the kvs from the mutation
table.addPendingUpdates(m.get(fam, qual));
@@ -224,24 +170,18 @@
Region region = Mockito.mock(Region.class);
Mockito.when(env.getRegion()).thenReturn(region);
- RegionScanner scanner = Mockito.mock(RegionScanner.class);
- Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenReturn(scanner);
final byte[] stored = Bytes.toBytes("stored-value");
final KeyValue storedKv = new KeyValue(row, fam, qual, ts, Type.Put, stored);
storedKv.setSequenceId(2);
- Mockito.when(scanner.next(Mockito.any(List.class))).thenAnswer(new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocation) throws Throwable {
- List<KeyValue> list = (List<KeyValue>) invocation.getArguments()[0];
- list.add(storedKv);
- return false;
- }
- });
- LocalHBaseState state = new LocalTable(env);
- LocalTableState table = new LocalTableState(state, m);
+ HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells =
+ new HashMap<ImmutableBytesPtr, List<Cell>>();
+ rowKeyPtrToCells.put(new ImmutableBytesPtr(row), Collections.singletonList((Cell)storedKv));
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(rowKeyPtrToCells);
+ LocalTableState table = new LocalTableState(cachedLocalTable, m);
+
// add the kvs from the mutation
- KeyValue kv = PhoenixKeyValueUtil.maybeCopyCell(m.get(fam, qual).get(0));
+ KeyValue kv = KeyValueUtil.ensureKeyValue(m.get(fam, qual).get(0));
kv.setSequenceId(0);
table.addPendingUpdates(kv);
@@ -258,8 +198,6 @@
p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
s = p.getFirst();
assertEquals("Didn't correctly rollback the row - still found it!", null, s.next());
- Mockito.verify(env, Mockito.times(1)).getRegion();
- Mockito.verify(region, Mockito.times(1)).getScanner(Mockito.any(Scan.class));
}
@SuppressWarnings("unchecked")
@@ -270,24 +208,19 @@
Region region = Mockito.mock(Region.class);
Mockito.when(env.getRegion()).thenReturn(region);
- RegionScanner scanner = Mockito.mock(RegionScanner.class);
- Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenReturn(scanner);
final KeyValue storedKv =
new KeyValue(row, fam, qual, ts, Type.Put, Bytes.toBytes("stored-value"));
storedKv.setSequenceId(2);
- Mockito.when(scanner.next(Mockito.any(List.class))).thenAnswer(new Answer<Boolean>() {
- @Override
- public Boolean answer(InvocationOnMock invocation) throws Throwable {
- List<KeyValue> list = (List<KeyValue>) invocation.getArguments()[0];
- list.add(storedKv);
- return false;
- }
- });
- LocalHBaseState state = new LocalTable(env);
+
Put pendingUpdate = new Put(row);
pendingUpdate.addColumn(fam, qual, ts, val);
- LocalTableState table = new LocalTableState(state, pendingUpdate);
+ HashMap<ImmutableBytesPtr, List<Cell>> rowKeyPtrToCells =
+ new HashMap<ImmutableBytesPtr, List<Cell>>();
+ rowKeyPtrToCells.put(new ImmutableBytesPtr(row), Collections.singletonList((Cell)storedKv));
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(rowKeyPtrToCells);
+ LocalTableState table = new LocalTableState(cachedLocalTable, pendingUpdate);
+
// do the lookup for the given column
ColumnReference col = new ColumnReference(fam, qual);
@@ -303,8 +236,6 @@
p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
s = p.getFirst();
assertEquals("Lost already loaded update!", storedKv, s.next());
- Mockito.verify(env, Mockito.times(1)).getRegion();
- Mockito.verify(region, Mockito.times(1)).getScanner(Mockito.any(Scan.class));
}
// TODO add test here for making sure multiple column references with the same column family don't
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilderTest.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilderTest.java
index f9c6798..dd67f90 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilderTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilderTest.java
@@ -51,7 +51,7 @@
import org.apache.phoenix.coprocessor.BaseRegionScanner;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver.ReplayWrite;
import org.apache.phoenix.hbase.index.MultiMutation;
-import org.apache.phoenix.hbase.index.covered.data.LocalTable;
+import org.apache.phoenix.hbase.index.covered.data.CachedLocalTable;
import org.apache.phoenix.hbase.index.covered.update.ColumnTracker;
import org.apache.phoenix.hbase.index.util.GenericKeyValueBuilder;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
@@ -149,6 +149,7 @@
mockIndexMetaData = Mockito.mock(PhoenixIndexMetaData.class);
Mockito.when(mockIndexMetaData.requiresPriorRowState((Mutation)Mockito.any())).thenReturn(true);
+ Mockito.when(mockIndexMetaData.getReplayWrite()).thenReturn(null);
Mockito.when(mockIndexMetaData.getIndexMaintainers())
.thenReturn(Collections.singletonList(getTestIndexMaintainer()));
@@ -212,8 +213,13 @@
MultiMutation mutation = new MultiMutation(new ImmutableBytesPtr(ROW));
mutation.addAll(put);
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(
+ Collections.singletonList(mutation),
+ this.mockIndexMetaData,
+ this.indexBuilder.getEnv().getRegion());
+
Collection<Pair<Mutation, byte[]>> indexUpdates =
- indexBuilder.getIndexUpdate(mutation, mockIndexMetaData);
+ indexBuilder.getIndexUpdate(mutation, mockIndexMetaData, cachedLocalTable);
assertEquals(2, indexUpdates.size());
assertContains(indexUpdates, 2, ROW, KeyValue.Type.DeleteFamily, FAM,
new byte[0] /* qual not needed */, 2);
@@ -254,8 +260,16 @@
mutation.addAll(put);
Collection<Pair<Mutation, byte[]>> indexUpdates = Lists.newArrayList();
- for (Mutation m : IndexManagementUtil.flattenMutationsByTimestamp(Collections.singletonList(mutation))) {
- indexUpdates.addAll(indexBuilder.getIndexUpdate(m, mockIndexMetaData));
+ Collection<? extends Mutation> mutations =
+ IndexManagementUtil.flattenMutationsByTimestamp(Collections.singletonList(mutation));
+
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(
+ mutations,
+ this.mockIndexMetaData,
+ this.indexBuilder.getEnv().getRegion());
+
+ for (Mutation m : mutations) {
+ indexUpdates.addAll(indexBuilder.getIndexUpdate(m, mockIndexMetaData, cachedLocalTable));
}
// 3 puts and 3 deletes (one to hide existing index row for VALUE_1, and two to hide index
// rows for VALUE_2, VALUE_3)
@@ -287,9 +301,17 @@
MultiMutation mutation = getMultipleVersionMutation(200);
currentRowCells = mutation.getFamilyCellMap().get(FAM);
+ Collection<? extends Mutation> mutations =
+ IndexManagementUtil.flattenMutationsByTimestamp(Collections.singletonList(mutation));
+
+ CachedLocalTable cachedLocalTable = CachedLocalTable.build(
+ mutations,
+ this.mockIndexMetaData,
+ this.indexBuilder.getEnv().getRegion());
+
Collection<Pair<Mutation, byte[]>> indexUpdates = Lists.newArrayList();
for (Mutation m : IndexManagementUtil.flattenMutationsByTimestamp(Collections.singletonList(mutation))) {
- indexUpdates.addAll(indexBuilder.getIndexUpdate(m, mockIndexMetaData));
+ indexUpdates.addAll(indexBuilder.getIndexUpdate(m, mockIndexMetaData, cachedLocalTable));
}
assertNotEquals(0, indexUpdates.size());
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/TestCoveredColumnIndexCodec.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/TestCoveredColumnIndexCodec.java
index 571ed85..8ed61e8 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/TestCoveredColumnIndexCodec.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/TestCoveredColumnIndexCodec.java
@@ -134,9 +134,9 @@
}
@Override
- public Result getCurrentRowState(Mutation m, Collection<? extends ColumnReference> toCover, boolean preMutationStateOnly)
+ public List<Cell> getCurrentRowState(Mutation m, Collection<? extends ColumnReference> toCover, boolean preMutationStateOnly)
throws IOException {
- return r;
+ return r.listCells();
}
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestLocalTable.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestLocalTable.java
deleted file mode 100644
index b11ac8d..0000000
--- a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/data/TestLocalTable.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.phoenix.hbase.index.covered.data;
-
-import static org.junit.Assert.assertEquals;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.KeyValue;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.junit.Test;
-
-public class TestLocalTable {
- private static final byte[] ROW = Bytes.toBytes("test_row");
-
- @Test
- public void testGetOldestTimestamp() {
- LocalTable localTable = new LocalTable(null);
-
- List<Cell> cellList1 = getCellList(new KeyValue(ROW, 5L), new KeyValue(ROW, 4L));
- assertEquals(4L, localTable.getOldestTimestamp(Collections.singletonList(cellList1)));
-
- List<Cell> cellList2 = getCellList(new KeyValue(ROW, 5L), new KeyValue(ROW, 2L));
- List<List<Cell>> set1 = new ArrayList<>(Arrays.asList(cellList1, cellList2));
- assertEquals(2L, localTable.getOldestTimestamp(set1));
-
- List<Cell> cellList3 = getCellList(new KeyValue(ROW, 1L));
- set1.add(cellList3);
- assertEquals(1L, localTable.getOldestTimestamp(set1));
-
- List<Cell> cellList4 =
- getCellList(new KeyValue(ROW, 3L), new KeyValue(ROW, 1L), new KeyValue(ROW, 0L));
- set1.add(cellList4);
- assertEquals(0L, localTable.getOldestTimestamp(set1));
- }
-
- private List<Cell> getCellList(KeyValue... kvs) {
- List<Cell> cellList = new ArrayList<>();
- for (KeyValue kv : kvs) {
- cellList.add(kv);
- }
- return cellList;
- }
-}
