| /* |
| * 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.tephra.hbase; |
| |
| import com.google.common.collect.ImmutableList; |
| import com.google.common.collect.Lists; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.hbase.Cell; |
| import org.apache.hadoop.hbase.CellUtil; |
| import org.apache.hadoop.hbase.Coprocessor; |
| import org.apache.hadoop.hbase.DoNotRetryIOException; |
| import org.apache.hadoop.hbase.HBaseTestingUtility; |
| import org.apache.hadoop.hbase.HColumnDescriptor; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.HTableDescriptor; |
| import org.apache.hadoop.hbase.KeyValue; |
| import org.apache.hadoop.hbase.TableName; |
| import org.apache.hadoop.hbase.client.Delete; |
| import org.apache.hadoop.hbase.client.Durability; |
| import org.apache.hadoop.hbase.client.Get; |
| import org.apache.hadoop.hbase.client.HBaseAdmin; |
| import org.apache.hadoop.hbase.client.HTable; |
| import org.apache.hadoop.hbase.client.HTableInterface; |
| import org.apache.hadoop.hbase.client.OperationWithAttributes; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.Result; |
| import org.apache.hadoop.hbase.client.ResultScanner; |
| import org.apache.hadoop.hbase.client.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.filter.CompareFilter; |
| import org.apache.hadoop.hbase.filter.LongComparator; |
| import org.apache.hadoop.hbase.filter.ValueFilter; |
| import org.apache.hadoop.hbase.regionserver.wal.WALEdit; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.tephra.Transaction; |
| import org.apache.tephra.TransactionConflictException; |
| import org.apache.tephra.TransactionContext; |
| import org.apache.tephra.TransactionManager; |
| import org.apache.tephra.TransactionSystemClient; |
| import org.apache.tephra.TxConstants; |
| import org.apache.tephra.hbase.coprocessor.TransactionProcessor; |
| import org.apache.tephra.inmemory.InMemoryTxSystemClient; |
| import org.apache.tephra.metrics.TxMetricsCollector; |
| import org.apache.tephra.persist.InMemoryTransactionStateStorage; |
| import org.apache.tephra.persist.TransactionStateStorage; |
| import org.junit.After; |
| import org.junit.AfterClass; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| |
| import static org.junit.Assert.assertArrayEquals; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| /** |
| * Tests for TransactionAwareHTables. |
| */ |
| public class TransactionAwareHTableTest { |
| private static final Logger LOG = LoggerFactory.getLogger(TransactionAwareHTableTest.class); |
| |
| private static HBaseTestingUtility testUtil; |
| private static HBaseAdmin hBaseAdmin; |
| private static TransactionStateStorage txStateStorage; |
| private static TransactionManager txManager; |
| private static Configuration conf; |
| private TransactionContext transactionContext; |
| private TransactionAwareHTable transactionAwareHTable; |
| private HTable hTable; |
| |
| private static final class TestBytes { |
| private static final byte[] table = Bytes.toBytes("testtable"); |
| private static final byte[] family = Bytes.toBytes("f1"); |
| private static final byte[] family2 = Bytes.toBytes("f2"); |
| private static final byte[] qualifier = Bytes.toBytes("col1"); |
| private static final byte[] qualifier2 = Bytes.toBytes("col2"); |
| private static final byte[] row = Bytes.toBytes("row"); |
| private static final byte[] row2 = Bytes.toBytes("row2"); |
| private static final byte[] row3 = Bytes.toBytes("row3"); |
| private static final byte[] row4 = Bytes.toBytes("row4"); |
| private static final byte[] value = Bytes.toBytes("value"); |
| private static final byte[] value2 = Bytes.toBytes("value2"); |
| private static final byte[] value3 = Bytes.toBytes("value3"); |
| } |
| |
| private static final String TEST_ATTRIBUTE = "TEST_ATTRIBUTE"; |
| |
| public static class TestRegionObserver extends BaseRegionObserver { |
| @Override |
| public void prePut(final ObserverContext<RegionCoprocessorEnvironment> c, |
| final Put put, final WALEdit edit, |
| final Durability durability) throws IOException { |
| if (put.getAttribute(TEST_ATTRIBUTE) == null) { |
| throw new DoNotRetryIOException("Put should preserve attributes"); |
| } |
| if (put.getDurability() != Durability.USE_DEFAULT) { |
| throw new DoNotRetryIOException("Durability is not propagated correctly"); |
| } |
| } |
| |
| @Override |
| public void preDelete(final ObserverContext<RegionCoprocessorEnvironment> c, |
| final Delete delete, final WALEdit edit, |
| final Durability durability) throws IOException { |
| if (delete.getAttribute(TEST_ATTRIBUTE) == null) { |
| throw new DoNotRetryIOException("Delete should preserve attributes"); |
| } |
| if (delete.getDurability() != Durability.USE_DEFAULT) { |
| throw new DoNotRetryIOException("Durability is not propagated correctly"); |
| } |
| } |
| } |
| |
| @BeforeClass |
| public static void setupBeforeClass() throws Exception { |
| testUtil = new HBaseTestingUtility(); |
| conf = testUtil.getConfiguration(); |
| |
| // Tune down the connection thread pool size |
| conf.setInt("hbase.hconnection.threads.core", 5); |
| conf.setInt("hbase.hconnection.threads.max", 10); |
| // Tunn down handler threads in regionserver |
| conf.setInt("hbase.regionserver.handler.count", 10); |
| |
| // Set to random port |
| conf.setInt("hbase.master.port", 0); |
| conf.setInt("hbase.master.info.port", 0); |
| conf.setInt("hbase.regionserver.port", 0); |
| conf.setInt("hbase.regionserver.info.port", 0); |
| |
| testUtil.startMiniCluster(); |
| hBaseAdmin = testUtil.getHBaseAdmin(); |
| txStateStorage = new InMemoryTransactionStateStorage(); |
| txManager = new TransactionManager(conf, txStateStorage, new TxMetricsCollector()); |
| txManager.startAndWait(); |
| } |
| |
| @AfterClass |
| public static void shutdownAfterClass() throws Exception { |
| testUtil.shutdownMiniCluster(); |
| hBaseAdmin.close(); |
| } |
| |
| @Before |
| public void setupBeforeTest() throws Exception { |
| hTable = createTable(TestBytes.table, new byte[][]{TestBytes.family}); |
| transactionAwareHTable = new TransactionAwareHTable(hTable); |
| transactionContext = new TransactionContext(new InMemoryTxSystemClient(txManager), transactionAwareHTable); |
| } |
| |
| @After |
| public void shutdownAfterTest() throws IOException { |
| hBaseAdmin.disableTable(TestBytes.table); |
| hBaseAdmin.deleteTable(TestBytes.table); |
| } |
| |
| private HTable createTable(byte[] tableName, byte[][] columnFamilies) throws Exception { |
| return createTable(tableName, columnFamilies, false, Collections.<String>emptyList()); |
| } |
| |
| private HTable createTable(byte[] tableName, byte[][] columnFamilies, boolean existingData, |
| List<String> coprocessors) throws Exception { |
| HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName)); |
| for (byte[] family : columnFamilies) { |
| HColumnDescriptor columnDesc = new HColumnDescriptor(family); |
| columnDesc.setMaxVersions(Integer.MAX_VALUE); |
| columnDesc.setValue(TxConstants.PROPERTY_TTL, String.valueOf(100000)); // in millis |
| desc.addFamily(columnDesc); |
| } |
| if (existingData) { |
| desc.setValue(TxConstants.READ_NON_TX_DATA, "true"); |
| } |
| // Divide individually to prevent any overflow |
| int priority = Coprocessor.PRIORITY_USER; |
| desc.addCoprocessor(TransactionProcessor.class.getName(), null, priority, null); |
| // order in list is the same order that coprocessors will be invoked |
| for (String coprocessor : coprocessors) { |
| desc.addCoprocessor(coprocessor, null, ++priority, null); |
| } |
| hBaseAdmin.createTable(desc); |
| testUtil.waitTableAvailable(tableName, 5000); |
| return new HTable(testUtil.getConfiguration(), tableName); |
| } |
| |
| /** |
| * Test transactional put and get requests. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testValidTransactionalPutAndGet() throws Exception { |
| transactionContext.start(); |
| Put put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| transactionAwareHTable.put(put); |
| transactionContext.finish(); |
| |
| transactionContext.start(); |
| Result result = transactionAwareHTable.get(new Get(TestBytes.row)); |
| transactionContext.finish(); |
| |
| byte[] value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value, value); |
| } |
| |
| /** |
| * Test aborted put requests, that must be rolled back. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testAbortedTransactionPutAndGet() throws Exception { |
| transactionContext.start(); |
| Put put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| transactionAwareHTable.put(put); |
| |
| transactionContext.abort(); |
| |
| transactionContext.start(); |
| Result result = transactionAwareHTable.get(new Get(TestBytes.row)); |
| transactionContext.finish(); |
| byte[] value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(value, null); |
| } |
| |
| /** |
| * Test transactional delete operations. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testValidTransactionalDelete() throws Exception { |
| HTable hTable = createTable(Bytes.toBytes("TestValidTransactionalDelete"), |
| new byte[][]{TestBytes.family, TestBytes.family2}); |
| try { |
| TransactionAwareHTable txTable = new TransactionAwareHTable(hTable); |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| |
| txContext.start(); |
| Put put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| put.add(TestBytes.family2, TestBytes.qualifier, TestBytes.value2); |
| txTable.put(put); |
| txContext.finish(); |
| |
| txContext.start(); |
| Result result = txTable.get(new Get(TestBytes.row)); |
| txContext.finish(); |
| byte[] value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value, value); |
| value = result.getValue(TestBytes.family2, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value2, value); |
| |
| // test full row delete |
| txContext.start(); |
| Delete delete = new Delete(TestBytes.row); |
| txTable.delete(delete); |
| txContext.finish(); |
| |
| txContext.start(); |
| result = txTable.get(new Get(TestBytes.row)); |
| txContext.finish(); |
| assertTrue(result.isEmpty()); |
| |
| // test column delete |
| // load 10 rows |
| txContext.start(); |
| int rowCount = 10; |
| for (int i = 0; i < rowCount; i++) { |
| Put p = new Put(Bytes.toBytes("row" + i)); |
| for (int j = 0; j < 10; j++) { |
| p.add(TestBytes.family, Bytes.toBytes(j), TestBytes.value); |
| } |
| txTable.put(p); |
| } |
| txContext.finish(); |
| |
| // verify loaded rows |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Get g = new Get(Bytes.toBytes("row" + i)); |
| Result r = txTable.get(g); |
| assertFalse(r.isEmpty()); |
| for (int j = 0; j < 10; j++) { |
| assertArrayEquals(TestBytes.value, r.getValue(TestBytes.family, Bytes.toBytes(j))); |
| } |
| } |
| txContext.finish(); |
| |
| // delete odds columns from odd rows and even columns from even rows |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Delete d = new Delete(Bytes.toBytes("row" + i)); |
| for (int j = 0; j < 10; j++) { |
| if (i % 2 == j % 2) { |
| LOG.info("Deleting row={}, column={}", i, j); |
| d.deleteColumns(TestBytes.family, Bytes.toBytes(j)); |
| } |
| } |
| txTable.delete(d); |
| } |
| txContext.finish(); |
| |
| // verify deleted columns |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Get g = new Get(Bytes.toBytes("row" + i)); |
| Result r = txTable.get(g); |
| assertEquals(5, r.size()); |
| for (Map.Entry<byte[], byte[]> entry : r.getFamilyMap(TestBytes.family).entrySet()) { |
| int col = Bytes.toInt(entry.getKey()); |
| LOG.info("Got row={}, col={}", i, col); |
| // each row should only have the opposite mod (odd=even, even=odd) |
| assertNotEquals(i % 2, col % 2); |
| assertArrayEquals(TestBytes.value, entry.getValue()); |
| } |
| } |
| txContext.finish(); |
| |
| // test family delete |
| // load 10 rows |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Put p = new Put(Bytes.toBytes("famrow" + i)); |
| p.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| p.add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value2); |
| txTable.put(p); |
| } |
| txContext.finish(); |
| |
| // verify all loaded rows |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Get g = new Get(Bytes.toBytes("famrow" + i)); |
| Result r = txTable.get(g); |
| assertEquals(2, r.size()); |
| assertArrayEquals(TestBytes.value, r.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value2, r.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| } |
| txContext.finish(); |
| |
| // delete family1 for even rows, family2 for odd rows |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Delete d = new Delete(Bytes.toBytes("famrow" + i)); |
| d.deleteFamily((i % 2 == 0) ? TestBytes.family : TestBytes.family2); |
| txTable.delete(d); |
| } |
| txContext.finish(); |
| |
| // verify deleted families |
| txContext.start(); |
| for (int i = 0; i < rowCount; i++) { |
| Get g = new Get(Bytes.toBytes("famrow" + i)); |
| Result r = txTable.get(g); |
| assertEquals(1, r.size()); |
| if (i % 2 == 0) { |
| assertNull(r.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value2, r.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| } else { |
| assertArrayEquals(TestBytes.value, r.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertNull(r.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| } |
| } |
| txContext.finish(); |
| } finally { |
| hTable.close(); |
| } |
| } |
| |
| /** |
| * Test that put and delete attributes are preserved |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testAttributesPreserved() throws Exception { |
| HTable hTable = createTable(Bytes.toBytes("TestAttributesPreserved"), |
| new byte[][]{TestBytes.family, TestBytes.family2}, false, |
| Lists.newArrayList(TestRegionObserver.class.getName())); |
| try { |
| TransactionAwareHTable txTable = new TransactionAwareHTable(hTable); |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| |
| txContext.start(); |
| Put put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| put.add(TestBytes.family2, TestBytes.qualifier, TestBytes.value2); |
| // set an attribute on the put, TestRegionObserver will verify it still exists |
| put.setAttribute(TEST_ATTRIBUTE, new byte[]{}); |
| txTable.put(put); |
| txContext.finish(); |
| |
| txContext.start(); |
| Result result = txTable.get(new Get(TestBytes.row)); |
| txContext.finish(); |
| byte[] value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value, value); |
| value = result.getValue(TestBytes.family2, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value2, value); |
| |
| // test full row delete, TestRegionObserver will verify it still exists |
| txContext.start(); |
| Delete delete = new Delete(TestBytes.row); |
| delete.setAttribute(TEST_ATTRIBUTE, new byte[]{}); |
| txTable.delete(delete); |
| txContext.finish(); |
| |
| txContext.start(); |
| result = txTable.get(new Get(TestBytes.row)); |
| txContext.finish(); |
| assertTrue(result.isEmpty()); |
| } finally { |
| hTable.close(); |
| } |
| } |
| |
| /** |
| * Test aborted transactional delete requests, that must be rolled back. |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testAbortedTransactionalDelete() throws Exception { |
| transactionContext.start(); |
| Put put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| transactionAwareHTable.put(put); |
| transactionContext.finish(); |
| |
| transactionContext.start(); |
| Result result = transactionAwareHTable.get(new Get(TestBytes.row)); |
| transactionContext.finish(); |
| byte[] value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value, value); |
| |
| transactionContext.start(); |
| Delete delete = new Delete(TestBytes.row); |
| transactionAwareHTable.delete(delete); |
| transactionContext.abort(); |
| |
| transactionContext.start(); |
| result = transactionAwareHTable.get(new Get(TestBytes.row)); |
| transactionContext.finish(); |
| value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value, value); |
| } |
| |
| private void testDeleteRollback(TxConstants.ConflictDetection conflictDetection) throws Exception { |
| String tableName = String.format("%s%s", "TestColFamilyDelete", conflictDetection); |
| HTable hTable = createTable(Bytes.toBytes(tableName), new byte[][]{TestBytes.family}); |
| try (TransactionAwareHTable txTable = new TransactionAwareHTable(hTable, conflictDetection)) { |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| txContext.finish(); |
| |
| // Start a tx, delete the row and then abort the tx |
| txContext.start(); |
| txTable.delete(new Delete(TestBytes.row)); |
| txContext.abort(); |
| |
| // Start a tx, delete a column family and then abort the tx |
| txContext.start(); |
| txTable.delete(new Delete(TestBytes.row).deleteFamily(TestBytes.family)); |
| txContext.abort(); |
| |
| // Above operations should have no effect on the row, since they were aborted |
| txContext.start(); |
| Get get = new Get(TestBytes.row); |
| Result result = txTable.get(get); |
| assertFalse(result.isEmpty()); |
| assertArrayEquals(TestBytes.value, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| txContext.finish(); |
| } |
| } |
| |
| @Test |
| public void testDeleteRollback() throws Exception { |
| testDeleteRollback(TxConstants.ConflictDetection.ROW); |
| testDeleteRollback(TxConstants.ConflictDetection.COLUMN); |
| testDeleteRollback(TxConstants.ConflictDetection.NONE); |
| } |
| |
| @Test |
| public void testMultiColumnFamilyRowDeleteRollback() throws Exception { |
| HTable hTable = createTable(Bytes.toBytes("TestMultColFam"), new byte[][] {TestBytes.family, TestBytes.family2}); |
| try (TransactionAwareHTable txTable = new TransactionAwareHTable(hTable, TxConstants.ConflictDetection.ROW)) { |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| txContext.finish(); |
| |
| txContext.start(); |
| //noinspection ConstantConditions |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_ALL); |
| Result result = txTable.get(new Get(TestBytes.row)); |
| Assert.assertEquals(1, result.getFamilyMap(TestBytes.family).size()); |
| Assert.assertEquals(0, result.getFamilyMap(TestBytes.family2).size()); |
| txContext.finish(); |
| |
| //Start a tx, delete the row and then abort the tx |
| txContext.start(); |
| txTable.delete(new Delete(TestBytes.row)); |
| txContext.abort(); |
| |
| //Start a tx and scan all the col families to make sure none of them have delete markers |
| txContext.start(); |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_ALL); |
| result = txTable.get(new Get(TestBytes.row)); |
| Assert.assertEquals(1, result.getFamilyMap(TestBytes.family).size()); |
| Assert.assertEquals(0, result.getFamilyMap(TestBytes.family2).size()); |
| txContext.finish(); |
| } |
| } |
| |
| @Test |
| public void testRowDelete() throws Exception { |
| HTable hTable = createTable(Bytes.toBytes("TestRowDelete"), new byte[][]{TestBytes.family, TestBytes.family2}); |
| TransactionAwareHTable txTable = new TransactionAwareHTable(hTable, TxConstants.ConflictDetection.ROW); |
| try { |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| |
| // Test 1: full row delete |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row) |
| .add(TestBytes.family, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family, TestBytes.qualifier2, TestBytes.value2) |
| .add(TestBytes.family2, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value2)); |
| txContext.finish(); |
| |
| txContext.start(); |
| Get get = new Get(TestBytes.row); |
| Result result = txTable.get(get); |
| assertFalse(result.isEmpty()); |
| assertArrayEquals(TestBytes.value, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value2, result.getValue(TestBytes.family, TestBytes.qualifier2)); |
| assertArrayEquals(TestBytes.value, result.getValue(TestBytes.family2, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value2, result.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| txContext.finish(); |
| |
| // delete entire row |
| txContext.start(); |
| txTable.delete(new Delete(TestBytes.row)); |
| txContext.finish(); |
| |
| // verify row is now empty |
| txContext.start(); |
| result = txTable.get(new Get(TestBytes.row)); |
| assertTrue(result.isEmpty()); |
| |
| // verify row is empty for explicit column retrieval |
| result = txTable.get(new Get(TestBytes.row) |
| .addColumn(TestBytes.family, TestBytes.qualifier) |
| .addFamily(TestBytes.family2)); |
| assertTrue(result.isEmpty()); |
| |
| // verify row is empty for scan |
| ResultScanner scanner = txTable.getScanner(new Scan(TestBytes.row)); |
| assertNull(scanner.next()); |
| scanner.close(); |
| |
| // verify row is empty for scan with explicit column |
| scanner = txTable.getScanner(new Scan(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier2)); |
| assertNull(scanner.next()); |
| scanner.close(); |
| txContext.finish(); |
| |
| // write swapped values to one column per family |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row) |
| .add(TestBytes.family, TestBytes.qualifier, TestBytes.value2) |
| .add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value)); |
| txContext.finish(); |
| |
| // verify new values appear |
| txContext.start(); |
| result = txTable.get(new Get(TestBytes.row)); |
| assertFalse(result.isEmpty()); |
| assertEquals(2, result.size()); |
| assertArrayEquals(TestBytes.value2, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value, result.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| |
| scanner = txTable.getScanner(new Scan(TestBytes.row)); |
| Result result1 = scanner.next(); |
| assertNotNull(result1); |
| assertFalse(result1.isEmpty()); |
| assertEquals(2, result1.size()); |
| assertArrayEquals(TestBytes.value2, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value, result.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| scanner.close(); |
| txContext.finish(); |
| |
| // Test 2: delete of first column family |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row2) |
| .add(TestBytes.family, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family, TestBytes.qualifier2, TestBytes.value2) |
| .add(TestBytes.family2, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value2)); |
| txContext.finish(); |
| |
| txContext.start(); |
| txTable.delete(new Delete(TestBytes.row2).deleteFamily(TestBytes.family)); |
| txContext.finish(); |
| |
| txContext.start(); |
| Result fam1Result = txTable.get(new Get(TestBytes.row2)); |
| assertFalse(fam1Result.isEmpty()); |
| assertEquals(2, fam1Result.size()); |
| assertArrayEquals(TestBytes.value, fam1Result.getValue(TestBytes.family2, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value2, fam1Result.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| txContext.finish(); |
| |
| // Test 3: delete of second column family |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row3) |
| .add(TestBytes.family, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family, TestBytes.qualifier2, TestBytes.value2) |
| .add(TestBytes.family2, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value2)); |
| txContext.finish(); |
| |
| txContext.start(); |
| txTable.delete(new Delete(TestBytes.row3).deleteFamily(TestBytes.family2)); |
| txContext.finish(); |
| |
| txContext.start(); |
| Result fam2Result = txTable.get(new Get(TestBytes.row3)); |
| assertFalse(fam2Result.isEmpty()); |
| assertEquals(2, fam2Result.size()); |
| assertArrayEquals(TestBytes.value, fam2Result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(TestBytes.value2, fam2Result.getValue(TestBytes.family, TestBytes.qualifier2)); |
| txContext.finish(); |
| |
| // Test 4: delete specific rows in a range |
| txContext.start(); |
| for (int i = 0; i < 10; i++) { |
| txTable.put(new Put(Bytes.toBytes("z" + i)) |
| .add(TestBytes.family, TestBytes.qualifier, Bytes.toBytes(i)) |
| .add(TestBytes.family2, TestBytes.qualifier2, Bytes.toBytes(i))); |
| } |
| txContext.finish(); |
| |
| txContext.start(); |
| // delete odd rows |
| for (int i = 1; i < 10; i += 2) { |
| txTable.delete(new Delete(Bytes.toBytes("z" + i))); |
| } |
| txContext.finish(); |
| |
| txContext.start(); |
| int cnt = 0; |
| ResultScanner zScanner = txTable.getScanner(new Scan(Bytes.toBytes("z0"))); |
| Result res; |
| while ((res = zScanner.next()) != null) { |
| assertFalse(res.isEmpty()); |
| assertArrayEquals(Bytes.toBytes("z" + cnt), res.getRow()); |
| assertArrayEquals(Bytes.toBytes(cnt), res.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(Bytes.toBytes(cnt), res.getValue(TestBytes.family2, TestBytes.qualifier2)); |
| cnt += 2; |
| } |
| |
| // Test 5: delete prior writes in the same transaction |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row4) |
| .add(TestBytes.family, TestBytes.qualifier, TestBytes.value) |
| .add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value2)); |
| txTable.delete(new Delete(TestBytes.row4)); |
| txContext.finish(); |
| |
| txContext.start(); |
| Result row4Result = txTable.get(new Get(TestBytes.row4)); |
| assertTrue(row4Result.isEmpty()); |
| txContext.finish(); |
| } finally { |
| txTable.close(); |
| } |
| } |
| |
| /** |
| * Expect an exception since a transaction hasn't been started. |
| * |
| * @throws Exception |
| */ |
| @Test(expected = IOException.class) |
| public void testTransactionlessFailure() throws Exception { |
| transactionAwareHTable.get(new Get(TestBytes.row)); |
| } |
| |
| /** |
| * Tests that each transaction can see its own persisted writes, while not seeing writes from other |
| * in-progress transactions. |
| */ |
| @Test |
| public void testReadYourWrites() throws Exception { |
| // In-progress tx1: started before our main transaction |
| HTable hTable1 = new HTable(testUtil.getConfiguration(), TestBytes.table); |
| TransactionAwareHTable txHTable1 = new TransactionAwareHTable(hTable1); |
| TransactionContext inprogressTxContext1 = new TransactionContext(new InMemoryTxSystemClient(txManager), txHTable1); |
| |
| // In-progress tx2: started while our main transaction is running |
| HTable hTable2 = new HTable(testUtil.getConfiguration(), TestBytes.table); |
| TransactionAwareHTable txHTable2 = new TransactionAwareHTable(hTable2); |
| TransactionContext inprogressTxContext2 = new TransactionContext(new InMemoryTxSystemClient(txManager), txHTable2); |
| |
| // create an in-progress write that should be ignored |
| byte[] col2 = Bytes.toBytes("col2"); |
| inprogressTxContext1.start(); |
| Put putCol2 = new Put(TestBytes.row); |
| byte[] valueCol2 = Bytes.toBytes("writing in progress"); |
| putCol2.add(TestBytes.family, col2, valueCol2); |
| txHTable1.put(putCol2); |
| |
| // start a tx and write a value to test reading in same tx |
| transactionContext.start(); |
| Put put = new Put(TestBytes.row); |
| byte[] value = Bytes.toBytes("writing"); |
| put.add(TestBytes.family, TestBytes.qualifier, value); |
| transactionAwareHTable.put(put); |
| |
| // test that a write from a tx started after the first is not visible |
| inprogressTxContext2.start(); |
| Put put2 = new Put(TestBytes.row); |
| byte[] value2 = Bytes.toBytes("writing2"); |
| put2.add(TestBytes.family, TestBytes.qualifier, value2); |
| txHTable2.put(put2); |
| |
| Get get = new Get(TestBytes.row); |
| Result row = transactionAwareHTable.get(get); |
| assertFalse(row.isEmpty()); |
| byte[] col1Value = row.getValue(TestBytes.family, TestBytes.qualifier); |
| Assert.assertNotNull(col1Value); |
| Assert.assertArrayEquals(value, col1Value); |
| // write from in-progress transaction should not be visible |
| byte[] col2Value = row.getValue(TestBytes.family, col2); |
| assertNull(col2Value); |
| |
| // commit in-progress transaction, should still not be visible |
| inprogressTxContext1.finish(); |
| |
| get = new Get(TestBytes.row); |
| row = transactionAwareHTable.get(get); |
| assertFalse(row.isEmpty()); |
| col2Value = row.getValue(TestBytes.family, col2); |
| assertNull(col2Value); |
| |
| transactionContext.finish(); |
| |
| inprogressTxContext2.abort(); |
| } |
| |
| @Test |
| public void testRowLevelConflictDetection() throws Exception { |
| TransactionAwareHTable txTable1 = new TransactionAwareHTable(new HTable(conf, TestBytes.table), |
| TxConstants.ConflictDetection.ROW); |
| TransactionContext txContext1 = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable1); |
| |
| TransactionAwareHTable txTable2 = new TransactionAwareHTable(new HTable(conf, TestBytes.table), |
| TxConstants.ConflictDetection.ROW); |
| TransactionContext txContext2 = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable2); |
| |
| byte[] row1 = Bytes.toBytes("row1"); |
| byte[] row2 = Bytes.toBytes("row2"); |
| byte[] col1 = Bytes.toBytes("c1"); |
| byte[] col2 = Bytes.toBytes("c2"); |
| byte[] val1 = Bytes.toBytes("val1"); |
| byte[] val2 = Bytes.toBytes("val2"); |
| |
| // test that concurrent writing to different rows succeeds |
| txContext1.start(); |
| txTable1.put(new Put(row1).add(TestBytes.family, col1, val1)); |
| |
| txContext2.start(); |
| txTable2.put(new Put(row2).add(TestBytes.family, col1, val2)); |
| |
| // should be no conflicts |
| txContext1.finish(); |
| txContext2.finish(); |
| |
| transactionContext.start(); |
| Result res = transactionAwareHTable.get(new Get(row1)); |
| assertFalse(res.isEmpty()); |
| Cell cell = res.getColumnLatestCell(TestBytes.family, col1); |
| assertNotNull(cell); |
| assertArrayEquals(val1, CellUtil.cloneValue(cell)); |
| |
| res = transactionAwareHTable.get(new Get(row2)); |
| assertFalse(res.isEmpty()); |
| cell = res.getColumnLatestCell(TestBytes.family, col1); |
| assertNotNull(cell); |
| assertArrayEquals(val2, CellUtil.cloneValue(cell)); |
| transactionContext.finish(); |
| |
| // test that writing to different columns in the same row fails |
| txContext1.start(); |
| txTable1.put(new Put(row1).add(TestBytes.family, col1, val2)); |
| |
| txContext2.start(); |
| txTable2.put(new Put(row1).add(TestBytes.family, col2, val2)); |
| |
| txContext1.finish(); |
| try { |
| txContext2.finish(); |
| fail("txContext2 should have encountered a row-level conflict during commit"); |
| } catch (TransactionConflictException tce) { |
| txContext2.abort(); |
| } |
| |
| transactionContext.start(); |
| res = transactionAwareHTable.get(new Get(row1)); |
| assertFalse(res.isEmpty()); |
| cell = res.getColumnLatestCell(TestBytes.family, col1); |
| assertNotNull(cell); |
| // should now be val2 |
| assertArrayEquals(val2, CellUtil.cloneValue(cell)); |
| |
| cell = res.getColumnLatestCell(TestBytes.family, col2); |
| // col2 should not be visible due to conflict |
| assertNull(cell); |
| transactionContext.finish(); |
| |
| // test that writing to the same column in the same row fails |
| txContext1.start(); |
| txTable1.put(new Put(row2).add(TestBytes.family, col2, val1)); |
| |
| txContext2.start(); |
| txTable2.put(new Put(row2).add(TestBytes.family, col2, val2)); |
| |
| txContext1.finish(); |
| try { |
| txContext2.finish(); |
| fail("txContext2 should have encountered a row and column level conflict during commit"); |
| } catch (TransactionConflictException tce) { |
| txContext2.abort(); |
| } |
| |
| transactionContext.start(); |
| res = transactionAwareHTable.get(new Get(row2)); |
| assertFalse(res.isEmpty()); |
| cell = res.getColumnLatestCell(TestBytes.family, col2); |
| assertNotNull(cell); |
| // should now be val1 |
| assertArrayEquals(val1, CellUtil.cloneValue(cell)); |
| transactionContext.finish(); |
| |
| // verify change set that is being reported only on rows |
| txContext1.start(); |
| txTable1.put(new Put(row1).add(TestBytes.family, col1, val1)); |
| txTable1.put(new Put(row2).add(TestBytes.family, col2, val2)); |
| |
| Collection<byte[]> changeSet = txTable1.getTxChanges(); |
| assertNotNull(changeSet); |
| assertEquals(2, changeSet.size()); |
| assertTrue(changeSet.contains(txTable1.getChangeKey(row1, null, null))); |
| assertTrue(changeSet.contains(txTable1.getChangeKey(row2, null, null))); |
| txContext1.finish(); |
| } |
| |
| @Test |
| public void testNoneLevelConflictDetection() throws Exception { |
| InMemoryTxSystemClient txClient = new InMemoryTxSystemClient(txManager); |
| TransactionAwareHTable txTable1 = new TransactionAwareHTable(new HTable(conf, TestBytes.table), |
| TxConstants.ConflictDetection.NONE); |
| TransactionContext txContext1 = new TransactionContext(txClient, txTable1); |
| |
| TransactionAwareHTable txTable2 = new TransactionAwareHTable(new HTable(conf, TestBytes.table), |
| TxConstants.ConflictDetection.NONE); |
| TransactionContext txContext2 = new TransactionContext(txClient, txTable2); |
| |
| // overlapping writes to the same row + column should not conflict |
| |
| txContext1.start(); |
| txTable1.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| |
| // changes should not be visible yet |
| txContext2.start(); |
| Result row = txTable2.get(new Get(TestBytes.row)); |
| assertTrue(row.isEmpty()); |
| |
| txTable2.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value2)); |
| |
| // both commits should succeed |
| txContext1.finish(); |
| txContext2.finish(); |
| |
| txContext1.start(); |
| row = txTable1.get(new Get(TestBytes.row)); |
| assertFalse(row.isEmpty()); |
| assertArrayEquals(TestBytes.value2, row.getValue(TestBytes.family, TestBytes.qualifier)); |
| txContext1.finish(); |
| |
| // transaction abort should still rollback changes |
| |
| txContext1.start(); |
| txTable1.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| txContext1.abort(); |
| |
| // changes to row2 should be rolled back |
| txContext2.start(); |
| Result row2 = txTable2.get(new Get(TestBytes.row2)); |
| assertTrue(row2.isEmpty()); |
| txContext2.finish(); |
| |
| // transaction invalidate should still make changes invisible |
| |
| txContext1.start(); |
| Transaction tx1 = txContext1.getCurrentTransaction(); |
| txTable1.put(new Put(TestBytes.row3).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| assertNotNull(tx1); |
| txClient.invalidate(tx1.getWritePointer()); |
| |
| // changes to row2 should be rolled back |
| txContext2.start(); |
| Result row3 = txTable2.get(new Get(TestBytes.row3)); |
| assertTrue(row3.isEmpty()); |
| txContext2.finish(); |
| } |
| |
| @Test |
| public void testCheckpoint() throws Exception { |
| // start a transaction, using checkpoints between writes |
| transactionContext.start(); |
| transactionAwareHTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| Transaction origTx = transactionContext.getCurrentTransaction(); |
| transactionContext.checkpoint(); |
| Transaction postCheckpointTx = transactionContext.getCurrentTransaction(); |
| |
| assertEquals(origTx.getTransactionId(), postCheckpointTx.getTransactionId()); |
| assertNotEquals(origTx.getWritePointer(), postCheckpointTx.getWritePointer()); |
| long[] checkpointPtrs = postCheckpointTx.getCheckpointWritePointers(); |
| assertEquals(1, checkpointPtrs.length); |
| assertEquals(postCheckpointTx.getWritePointer(), checkpointPtrs[0]); |
| |
| transactionAwareHTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value2)); |
| transactionContext.checkpoint(); |
| Transaction postCheckpointTx2 = transactionContext.getCurrentTransaction(); |
| |
| assertEquals(origTx.getTransactionId(), postCheckpointTx2.getTransactionId()); |
| assertNotEquals(postCheckpointTx.getWritePointer(), postCheckpointTx2.getWritePointer()); |
| long[] checkpointPtrs2 = postCheckpointTx2.getCheckpointWritePointers(); |
| assertEquals(2, checkpointPtrs2.length); |
| assertEquals(postCheckpointTx.getWritePointer(), checkpointPtrs2[0]); |
| assertEquals(postCheckpointTx2.getWritePointer(), checkpointPtrs2[1]); |
| |
| transactionAwareHTable.put(new Put(TestBytes.row3).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| |
| // by default, all rows should be visible with Read-Your-Writes |
| verifyRow(transactionAwareHTable, TestBytes.row, TestBytes.value); |
| verifyRow(transactionAwareHTable, TestBytes.row2, TestBytes.value2); |
| verifyRow(transactionAwareHTable, TestBytes.row3, TestBytes.value); |
| |
| // when disabling current write pointer, only the previous checkpoints should be visible |
| transactionContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_EXCLUDE_CURRENT); |
| Get get = new Get(TestBytes.row); |
| verifyRow(transactionAwareHTable, get, TestBytes.value); |
| get = new Get(TestBytes.row2); |
| verifyRow(transactionAwareHTable, get, TestBytes.value2); |
| get = new Get(TestBytes.row3); |
| verifyRow(transactionAwareHTable, get, null); |
| |
| // test scan results excluding current write pointer |
| Scan scan = new Scan(); |
| ResultScanner scanner = transactionAwareHTable.getScanner(scan); |
| |
| Result row = scanner.next(); |
| assertNotNull(row); |
| assertArrayEquals(TestBytes.row, row.getRow()); |
| assertEquals(1, row.size()); |
| assertArrayEquals(TestBytes.value, row.getValue(TestBytes.family, TestBytes.qualifier)); |
| |
| row = scanner.next(); |
| assertNotNull(row); |
| assertArrayEquals(TestBytes.row2, row.getRow()); |
| assertEquals(1, row.size()); |
| assertArrayEquals(TestBytes.value2, row.getValue(TestBytes.family, TestBytes.qualifier)); |
| |
| row = scanner.next(); |
| assertNull(row); |
| scanner.close(); |
| transactionContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT); |
| |
| // commit transaction, verify writes are visible |
| transactionContext.finish(); |
| |
| transactionContext.start(); |
| verifyRow(transactionAwareHTable, TestBytes.row, TestBytes.value); |
| verifyRow(transactionAwareHTable, TestBytes.row2, TestBytes.value2); |
| verifyRow(transactionAwareHTable, TestBytes.row3, TestBytes.value); |
| transactionContext.finish(); |
| } |
| |
| @Test |
| public void testInProgressCheckpoint() throws Exception { |
| // start a transaction, using checkpoints between writes |
| transactionContext.start(); |
| transactionAwareHTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| transactionContext.checkpoint(); |
| transactionAwareHTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value2)); |
| |
| // check that writes are still not visible to other clients |
| TransactionAwareHTable txTable2 = new TransactionAwareHTable(new HTable(conf, TestBytes.table)); |
| TransactionContext txContext2 = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable2); |
| |
| txContext2.start(); |
| verifyRow(txTable2, TestBytes.row, null); |
| verifyRow(txTable2, TestBytes.row2, null); |
| txContext2.finish(); |
| txTable2.close(); |
| |
| transactionContext.finish(); |
| |
| // verify writes are visible after commit |
| transactionContext.start(); |
| verifyRow(transactionAwareHTable, TestBytes.row, TestBytes.value); |
| verifyRow(transactionAwareHTable, TestBytes.row2, TestBytes.value2); |
| transactionContext.finish(); |
| } |
| |
| @Test |
| public void testCheckpointRollback() throws Exception { |
| // start a transaction, using checkpoints between writes |
| transactionContext.start(); |
| transactionAwareHTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| transactionContext.checkpoint(); |
| transactionAwareHTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value2)); |
| transactionContext.checkpoint(); |
| transactionAwareHTable.put(new Put(TestBytes.row3).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| |
| transactionContext.abort(); |
| |
| transactionContext.start(); |
| verifyRow(transactionAwareHTable, TestBytes.row, null); |
| verifyRow(transactionAwareHTable, TestBytes.row2, null); |
| verifyRow(transactionAwareHTable, TestBytes.row3, null); |
| |
| Scan scan = new Scan(); |
| ResultScanner scanner = transactionAwareHTable.getScanner(scan); |
| assertNull(scanner.next()); |
| scanner.close(); |
| transactionContext.finish(); |
| } |
| |
| @Test |
| public void testCheckpointInvalidate() throws Exception { |
| // start a transaction, using checkpoints between writes |
| transactionContext.start(); |
| Transaction origTx = transactionContext.getCurrentTransaction(); |
| transactionAwareHTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| transactionContext.checkpoint(); |
| Transaction checkpointTx1 = transactionContext.getCurrentTransaction(); |
| transactionAwareHTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value2)); |
| transactionContext.checkpoint(); |
| Transaction checkpointTx2 = transactionContext.getCurrentTransaction(); |
| transactionAwareHTable.put(new Put(TestBytes.row3).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| |
| TransactionSystemClient txClient = new InMemoryTxSystemClient(txManager); |
| txClient.invalidate(transactionContext.getCurrentTransaction().getTransactionId()); |
| |
| // check that writes are not visible |
| TransactionAwareHTable txTable2 = new TransactionAwareHTable(new HTable(conf, TestBytes.table)); |
| TransactionContext txContext2 = new TransactionContext(txClient, txTable2); |
| txContext2.start(); |
| Transaction newTx = txContext2.getCurrentTransaction(); |
| |
| // all 3 writes pointers from the previous transaction should now be excluded |
| assertTrue(newTx.isExcluded(origTx.getWritePointer())); |
| assertTrue(newTx.isExcluded(checkpointTx1.getWritePointer())); |
| assertTrue(newTx.isExcluded(checkpointTx2.getWritePointer())); |
| |
| verifyRow(txTable2, TestBytes.row, null); |
| verifyRow(txTable2, TestBytes.row2, null); |
| verifyRow(txTable2, TestBytes.row3, null); |
| |
| Scan scan = new Scan(); |
| ResultScanner scanner = txTable2.getScanner(scan); |
| assertNull(scanner.next()); |
| scanner.close(); |
| txContext2.finish(); |
| } |
| |
| @Test |
| public void testExistingData() throws Exception { |
| byte[] val11 = Bytes.toBytes("val11"); |
| byte[] val12 = Bytes.toBytes("val12"); |
| byte[] val21 = Bytes.toBytes("val21"); |
| byte[] val22 = Bytes.toBytes("val22"); |
| byte[] val31 = Bytes.toBytes("val31"); |
| byte[] val111 = Bytes.toBytes("val111"); |
| |
| TransactionAwareHTable txTable = |
| new TransactionAwareHTable(createTable(Bytes.toBytes("testExistingData"), new byte[][]{TestBytes.family}, true, |
| Collections.<String>emptyList())); |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| |
| // Add some pre-existing, non-transactional data |
| HTable nonTxTable = new HTable(testUtil.getConfiguration(), txTable.getTableName()); |
| nonTxTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, val11)); |
| nonTxTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier2, val12)); |
| nonTxTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, val21)); |
| nonTxTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier2, val22)); |
| nonTxTable.put(new Put(TestBytes.row4).add(TestBytes.family, TxConstants.FAMILY_DELETE_QUALIFIER, |
| HConstants.EMPTY_BYTE_ARRAY)); |
| nonTxTable.put(new Put(TestBytes.row4).add(TestBytes.family, TestBytes.qualifier, HConstants.EMPTY_BYTE_ARRAY)); |
| nonTxTable.flushCommits(); |
| |
| // Add transactional data |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row3).add(TestBytes.family, TestBytes.qualifier, val31)); |
| txContext.finish(); |
| |
| txContext.start(); |
| // test get |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier), val11); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier2), val12); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier), val21); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier2), val22); |
| verifyRow(txTable, new Get(TestBytes.row3).addColumn(TestBytes.family, TestBytes.qualifier), val31); |
| verifyRow(txTable, new Get(TestBytes.row4).addColumn(TestBytes.family, TxConstants.FAMILY_DELETE_QUALIFIER), |
| HConstants.EMPTY_BYTE_ARRAY); |
| verifyRow(txTable, new Get(TestBytes.row4).addColumn(TestBytes.family, TestBytes.qualifier), |
| HConstants.EMPTY_BYTE_ARRAY); |
| |
| // test scan |
| try (ResultScanner scanner = txTable.getScanner(new Scan())) { |
| Result result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row, result.getRow()); |
| assertArrayEquals(val11, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(val12, result.getValue(TestBytes.family, TestBytes.qualifier2)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row2, result.getRow()); |
| assertArrayEquals(val21, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(val22, result.getValue(TestBytes.family, TestBytes.qualifier2)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row3, result.getRow()); |
| assertArrayEquals(val31, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row4, result.getRow()); |
| assertArrayEquals(HConstants.EMPTY_BYTE_ARRAY, result.getValue(TestBytes.family, |
| TxConstants.FAMILY_DELETE_QUALIFIER)); |
| assertArrayEquals(HConstants.EMPTY_BYTE_ARRAY, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertNull(scanner.next()); |
| } |
| txContext.finish(); |
| |
| // test update and delete |
| txContext.start(); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, val111)); |
| txTable.delete(new Delete(TestBytes.row2).deleteColumns(TestBytes.family, TestBytes.qualifier)); |
| txContext.finish(); |
| |
| txContext.start(); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier), val111); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier2), val12); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier), null); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier2), val22); |
| verifyRow(txTable, new Get(TestBytes.row3).addColumn(TestBytes.family, TestBytes.qualifier), val31); |
| verifyRow(txTable, new Get(TestBytes.row4).addColumn(TestBytes.family, TxConstants.FAMILY_DELETE_QUALIFIER), |
| HConstants.EMPTY_BYTE_ARRAY); |
| verifyRow(txTable, new Get(TestBytes.row4).addColumn(TestBytes.family, TestBytes.qualifier), |
| HConstants.EMPTY_BYTE_ARRAY); |
| txContext.finish(); |
| |
| // test scan |
| txContext.start(); |
| try (ResultScanner scanner = txTable.getScanner(new Scan())) { |
| Result result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row, result.getRow()); |
| assertArrayEquals(val111, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(val12, result.getValue(TestBytes.family, TestBytes.qualifier2)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row2, result.getRow()); |
| assertArrayEquals(null, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertArrayEquals(val22, result.getValue(TestBytes.family, TestBytes.qualifier2)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row3, result.getRow()); |
| assertArrayEquals(val31, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row4, result.getRow()); |
| assertArrayEquals(HConstants.EMPTY_BYTE_ARRAY, result.getValue(TestBytes.family, |
| TxConstants.FAMILY_DELETE_QUALIFIER)); |
| assertArrayEquals(HConstants.EMPTY_BYTE_ARRAY, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| assertNull(scanner.next()); |
| } |
| txContext.finish(); |
| } |
| |
| private void verifyRow(HTableInterface table, byte[] rowkey, byte[] expectedValue) throws Exception { |
| verifyRow(table, new Get(rowkey), expectedValue); |
| } |
| |
| private void verifyRow(HTableInterface table, Get get, byte[] expectedValue) throws Exception { |
| verifyRows(table, get, expectedValue == null ? null : ImmutableList.of(expectedValue)); |
| } |
| |
| private void verifyRows(HTableInterface table, Get get, List<byte[]> expectedValues) throws Exception { |
| Result result = table.get(get); |
| if (expectedValues == null) { |
| assertTrue(result.isEmpty()); |
| } else { |
| assertFalse(result.isEmpty()); |
| byte[] family = TestBytes.family; |
| byte[] col = TestBytes.qualifier; |
| if (get.hasFamilies()) { |
| family = get.getFamilyMap().keySet().iterator().next(); |
| col = get.getFamilyMap().get(family).first(); |
| } |
| Iterator<Cell> it = result.getColumnCells(family, col).iterator(); |
| for (byte[] expectedValue : expectedValues) { |
| Assert.assertTrue(it.hasNext()); |
| assertArrayEquals(expectedValue, CellUtil.cloneValue(it.next())); |
| } |
| } |
| } |
| |
| private Cell[] getRow(HTableInterface table, Get get) throws Exception { |
| Result result = table.get(get); |
| return result.rawCells(); |
| } |
| |
| private void verifyScan(HTableInterface table, Scan scan, List<KeyValue> expectedCells) throws Exception { |
| List<Cell> actualCells = new ArrayList<>(); |
| try (ResultScanner scanner = table.getScanner(scan)) { |
| Result[] results = scanner.next(expectedCells.size() + 1); |
| for (Result result : results) { |
| actualCells.addAll(Lists.newArrayList(result.rawCells())); |
| } |
| Assert.assertEquals(expectedCells, actualCells); |
| } |
| } |
| |
| @Test |
| public void testVisibilityAll() throws Exception { |
| HTable nonTxTable = createTable(Bytes.toBytes("testVisibilityAll"), |
| new byte[][]{TestBytes.family, TestBytes.family2}, true, Collections.<String>emptyList()); |
| TransactionAwareHTable txTable = |
| new TransactionAwareHTable(nonTxTable, |
| TxConstants.ConflictDetection.ROW); // ROW conflict detection to verify family deletes |
| TransactionContext txContext = new TransactionContext(new InMemoryTxSystemClient(txManager), txTable); |
| |
| // start a transaction and create a delete marker |
| txContext.start(); |
| //noinspection ConstantConditions |
| long txWp0 = txContext.getCurrentTransaction().getWritePointer(); |
| txTable.delete(new Delete(TestBytes.row).deleteColumn(TestBytes.family, TestBytes.qualifier2)); |
| txContext.finish(); |
| |
| // start a new transaction and write some values |
| txContext.start(); |
| @SuppressWarnings("ConstantConditions") |
| long txWp1 = txContext.getCurrentTransaction().getWritePointer(); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier2, TestBytes.value2)); |
| txTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value)); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family2, TestBytes.qualifier, TestBytes.value)); |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family2, TestBytes.qualifier2, TestBytes.value2)); |
| |
| // verify written data |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier), |
| TestBytes.value); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier2), |
| TestBytes.value2); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier), |
| TestBytes.value); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier), |
| TestBytes.value); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier2), |
| TestBytes.value2); |
| |
| // checkpoint and make changes to written data now |
| txContext.checkpoint(); |
| long txWp2 = txContext.getCurrentTransaction().getWritePointer(); |
| // delete a column |
| txTable.delete(new Delete(TestBytes.row).deleteColumn(TestBytes.family, TestBytes.qualifier)); |
| // no change to a column |
| txTable.put(new Put(TestBytes.row).add(TestBytes.family, TestBytes.qualifier2, TestBytes.value2)); |
| // update a column |
| txTable.put(new Put(TestBytes.row2).add(TestBytes.family, TestBytes.qualifier, TestBytes.value3)); |
| // delete column family |
| txTable.delete(new Delete(TestBytes.row).deleteFamily(TestBytes.family2)); |
| |
| // verify changed values |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier), |
| null); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier2), |
| TestBytes.value2); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier), |
| TestBytes.value3); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier), |
| null); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier2), |
| null); |
| |
| // run a scan with VisibilityLevel.ALL, this should return all raw changes by this transaction, |
| // and the raw change by prior transaction |
| //noinspection ConstantConditions |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_ALL); |
| List<KeyValue> expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier, txWp2, KeyValue.Type.DeleteColumn), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier, txWp1, TestBytes.value), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp2, TestBytes.value2), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp1, TestBytes.value2), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp0, KeyValue.Type.DeleteColumn), |
| new KeyValue(TestBytes.row, TestBytes.family2, null, txWp2, KeyValue.Type.DeleteFamily), |
| new KeyValue(TestBytes.row, TestBytes.family2, TestBytes.qualifier, txWp1, TestBytes.value), |
| new KeyValue(TestBytes.row, TestBytes.family2, TestBytes.qualifier2, txWp1, TestBytes.value2), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp2, TestBytes.value3), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp1, TestBytes.value) |
| ); |
| verifyScan(txTable, new Scan(), expected); |
| |
| // verify a Get is also able to return all snapshot versions |
| Get get = new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier); |
| Cell[] cells = getRow(txTable, get); |
| Assert.assertEquals(2, cells.length); |
| Assert.assertTrue(CellUtil.isDelete(cells[0])); |
| Assert.assertArrayEquals(TestBytes.value, CellUtil.cloneValue(cells[1])); |
| |
| get = new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier2); |
| cells = getRow(txTable, get); |
| Assert.assertEquals(3, cells.length); |
| Assert.assertArrayEquals(TestBytes.value2, CellUtil.cloneValue(cells[0])); |
| Assert.assertArrayEquals(TestBytes.value2, CellUtil.cloneValue(cells[1])); |
| Assert.assertTrue(CellUtil.isDeleteColumns(cells[2])); |
| |
| verifyRows(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier), |
| ImmutableList.of(TestBytes.value3, TestBytes.value)); |
| |
| get = new Get(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier); |
| cells = getRow(txTable, get); |
| Assert.assertEquals(2, cells.length); |
| Assert.assertTrue(CellUtil.isDelete(cells[0])); |
| Assert.assertArrayEquals(TestBytes.value, CellUtil.cloneValue(cells[1])); |
| |
| get = new Get(TestBytes.row).addColumn(TestBytes.family2, TestBytes.qualifier2); |
| cells = getRow(txTable, get); |
| Assert.assertEquals(2, cells.length); |
| Assert.assertTrue(CellUtil.isDelete(cells[0])); |
| Assert.assertArrayEquals(TestBytes.value2, CellUtil.cloneValue(cells[1])); |
| |
| // Verify VisibilityLevel.SNAPSHOT |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT); |
| expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp2, TestBytes.value2), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp2, TestBytes.value3) |
| ); |
| verifyScan(txTable, new Scan(), expected); |
| |
| // Verify VisibilityLevel.SNAPSHOT_EXCLUDE_CURRENT |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_EXCLUDE_CURRENT); |
| expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier, txWp1, TestBytes.value), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp1, TestBytes.value2), |
| new KeyValue(TestBytes.row, TestBytes.family2, TestBytes.qualifier, txWp1, TestBytes.value), |
| new KeyValue(TestBytes.row, TestBytes.family2, TestBytes.qualifier2, txWp1, TestBytes.value2), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp1, TestBytes.value) |
| ); |
| verifyScan(txTable, new Scan(), expected); |
| txContext.finish(); |
| |
| // finally verify values once more after commit, this time we should get only committed raw values for |
| // all visibility levels |
| txContext.start(); |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_ALL); |
| expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier, txWp2, KeyValue.Type.DeleteColumn), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp2, TestBytes.value2), |
| new KeyValue(TestBytes.row, TestBytes.family2, null, txWp2, KeyValue.Type.DeleteFamily), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp2, TestBytes.value3) |
| ); |
| verifyScan(txTable, new Scan(), expected); |
| |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT); |
| expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp2, TestBytes.value2), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp2, TestBytes.value3) |
| ); |
| verifyScan(txTable, new Scan(), expected); |
| |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_EXCLUDE_CURRENT); |
| expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp2, TestBytes.value2), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp2, TestBytes.value3) |
| ); |
| verifyScan(txTable, new Scan(), expected); |
| |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier), |
| null); |
| verifyRow(txTable, new Get(TestBytes.row).addColumn(TestBytes.family, TestBytes.qualifier2), |
| TestBytes.value2); |
| verifyRow(txTable, new Get(TestBytes.row2).addColumn(TestBytes.family, TestBytes.qualifier), |
| TestBytes.value3); |
| txContext.finish(); |
| |
| // Test with regular HBase deletes in pre-existing data |
| long now = System.currentTimeMillis(); |
| Delete deleteColumn = new Delete(TestBytes.row3).deleteColumn(TestBytes.family, TestBytes.qualifier, now - 1); |
| // to prevent Tephra from replacing delete with delete marker |
| deleteColumn.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]); |
| nonTxTable.delete(deleteColumn); |
| Delete deleteFamily = new Delete(TestBytes.row3).deleteFamily(TestBytes.family2, now); |
| // to prevent Tephra from replacing delete with delete marker |
| deleteFamily.setAttribute(TxConstants.TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]); |
| nonTxTable.delete(deleteFamily); |
| nonTxTable.flushCommits(); |
| |
| txContext.start(); |
| txContext.getCurrentTransaction().setVisibility(Transaction.VisibilityLevel.SNAPSHOT_ALL); |
| expected = ImmutableList.of( |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier, txWp2, KeyValue.Type.DeleteColumn), |
| new KeyValue(TestBytes.row, TestBytes.family, TestBytes.qualifier2, txWp2, TestBytes.value2), |
| new KeyValue(TestBytes.row, TestBytes.family2, null, txWp2, KeyValue.Type.DeleteFamily), |
| new KeyValue(TestBytes.row2, TestBytes.family, TestBytes.qualifier, txWp2, TestBytes.value3), |
| new KeyValue(TestBytes.row3, TestBytes.family, TestBytes.qualifier, now - 1, KeyValue.Type.Delete), |
| new KeyValue(TestBytes.row3, TestBytes.family2, null, now, KeyValue.Type.DeleteFamily) |
| ); |
| // test scan |
| Scan scan = new Scan(); |
| scan.setRaw(true); |
| verifyScan(txTable, scan, expected); |
| txContext.finish(); |
| } |
| |
| @Test |
| public void testFilters() throws Exception { |
| // Add some values to table |
| transactionContext.start(); |
| Put put = new Put(TestBytes.row); |
| byte[] val1 = Bytes.toBytes(1L); |
| put.add(TestBytes.family, TestBytes.qualifier, val1); |
| transactionAwareHTable.put(put); |
| put = new Put(TestBytes.row2); |
| byte[] val2 = Bytes.toBytes(2L); |
| put.add(TestBytes.family, TestBytes.qualifier, val2); |
| transactionAwareHTable.put(put); |
| put = new Put(TestBytes.row3); |
| byte[] val3 = Bytes.toBytes(3L); |
| put.add(TestBytes.family, TestBytes.qualifier, val3); |
| transactionAwareHTable.put(put); |
| put = new Put(TestBytes.row4); |
| byte[] val4 = Bytes.toBytes(4L); |
| put.add(TestBytes.family, TestBytes.qualifier, val4); |
| transactionAwareHTable.put(put); |
| transactionContext.finish(); |
| |
| // Delete cell with value 2 |
| transactionContext.start(); |
| Delete delete = new Delete(TestBytes.row2); |
| delete.addColumn(TestBytes.family, TestBytes.qualifier); |
| transactionAwareHTable.delete(delete); |
| transactionContext.finish(); |
| |
| // Scan for values less than 4, should get only values 1 and 3 |
| transactionContext.start(); |
| Scan scan = new Scan(TestBytes.row, new ValueFilter(CompareFilter.CompareOp.LESS, new LongComparator(4))); |
| try (ResultScanner scanner = transactionAwareHTable.getScanner(scan)) { |
| Result result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row, result.getRow()); |
| assertArrayEquals(val1, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row3, result.getRow()); |
| assertArrayEquals(val3, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| result = scanner.next(); |
| assertNull(result); |
| } |
| transactionContext.finish(); |
| |
| // Run a Get with a filter for less than 10 on row4, should get value 4 |
| transactionContext.start(); |
| Get get = new Get(TestBytes.row4); |
| get.setFilter(new ValueFilter(CompareFilter.CompareOp.LESS, new LongComparator(10))); |
| Result result = transactionAwareHTable.get(get); |
| assertFalse(result.isEmpty()); |
| assertArrayEquals(val4, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| transactionContext.finish(); |
| |
| // Change value of row4 to 40 |
| transactionContext.start(); |
| put = new Put(TestBytes.row4); |
| byte[] val40 = Bytes.toBytes(40L); |
| put.add(TestBytes.family, TestBytes.qualifier, val40); |
| transactionAwareHTable.put(put); |
| transactionContext.finish(); |
| |
| // Scan for values less than 10, should get only values 1 and 3 |
| transactionContext.start(); |
| scan = new Scan(TestBytes.row, new ValueFilter(CompareFilter.CompareOp.LESS, new LongComparator(10))); |
| try (ResultScanner scanner = transactionAwareHTable.getScanner(scan)) { |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row, result.getRow()); |
| assertArrayEquals(val1, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| result = scanner.next(); |
| assertNotNull(result); |
| assertArrayEquals(TestBytes.row3, result.getRow()); |
| assertArrayEquals(val3, result.getValue(TestBytes.family, TestBytes.qualifier)); |
| result = scanner.next(); |
| assertNull(result); |
| } |
| transactionContext.finish(); |
| |
| // Run the Get again with a filter for less than 10 on row4, this time should not get any results |
| transactionContext.start(); |
| result = transactionAwareHTable.get(get); |
| assertTrue(result.isEmpty()); |
| transactionContext.finish(); |
| } |
| |
| /** |
| * Tests that transaction co-processor works with older clients |
| * |
| * @throws Exception |
| */ |
| @Test |
| public void testOlderClientOperations() throws Exception { |
| // Use old HTable to test |
| TransactionAwareHTable oldTxAware = new OldTransactionAwareHTable(hTable); |
| transactionContext.addTransactionAware(oldTxAware); |
| |
| transactionContext.start(); |
| Put put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value); |
| oldTxAware.put(put); |
| transactionContext.finish(); |
| |
| transactionContext.start(); |
| long txId = transactionContext.getCurrentTransaction().getTransactionId(); |
| put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value2); |
| oldTxAware.put(put); |
| // Invalidate the second Put |
| TransactionSystemClient txClient = new InMemoryTxSystemClient(txManager); |
| txClient.invalidate(txId); |
| |
| transactionContext.start(); |
| put = new Put(TestBytes.row); |
| put.add(TestBytes.family, TestBytes.qualifier, TestBytes.value3); |
| oldTxAware.put(put); |
| // Abort the third Put |
| transactionContext.abort(); |
| |
| // Get should now return the first value |
| transactionContext.start(); |
| Result result = oldTxAware.get(new Get(TestBytes.row)); |
| transactionContext.finish(); |
| |
| byte[] value = result.getValue(TestBytes.family, TestBytes.qualifier); |
| assertArrayEquals(TestBytes.value, value); |
| } |
| |
| /** |
| * Represents older transaction clients |
| */ |
| private static class OldTransactionAwareHTable extends TransactionAwareHTable { |
| public OldTransactionAwareHTable(HTableInterface hTable) { |
| super(hTable); |
| } |
| |
| @Override |
| public void addToOperation(OperationWithAttributes op, Transaction tx) throws IOException { |
| op.setAttribute(TxConstants.OLD_TX_OPERATION_ATTRIBUTE_KEY, txCodec.encode(tx)); |
| } |
| |
| @Override |
| protected void makeRollbackOperation(Delete delete) { |
| delete.setAttribute(TxConstants.OLD_TX_ROLLBACK_ATTRIBUTE_KEY, new byte[0]); |
| } |
| } |
| } |
