001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.client; 019 020import static org.junit.jupiter.api.Assertions.assertEquals; 021import static org.junit.jupiter.api.Assertions.assertTrue; 022 023import java.io.IOException; 024import java.util.ArrayList; 025import java.util.Arrays; 026import java.util.List; 027import org.apache.hadoop.hbase.Cell; 028import org.apache.hadoop.hbase.CellUtil; 029import org.apache.hadoop.hbase.HBaseTestingUtil; 030import org.apache.hadoop.hbase.TableName; 031import org.apache.hadoop.hbase.filter.Filter; 032import org.apache.hadoop.hbase.filter.TimestampsFilter; 033import org.apache.hadoop.hbase.testclassification.ClientTests; 034import org.apache.hadoop.hbase.testclassification.MediumTests; 035import org.apache.hadoop.hbase.util.Bytes; 036import org.junit.jupiter.api.AfterAll; 037import org.junit.jupiter.api.AfterEach; 038import org.junit.jupiter.api.BeforeAll; 039import org.junit.jupiter.api.BeforeEach; 040import org.junit.jupiter.api.Tag; 041import org.junit.jupiter.api.Test; 042import org.junit.jupiter.api.TestInfo; 043import org.slf4j.Logger; 044import org.slf4j.LoggerFactory; 045 046/** 047 * Run tests related to {@link TimestampsFilter} using HBase client APIs. Sets up the HBase mini 048 * cluster once at start. Each creates a table named for the method and does its stuff against that. 049 */ 050@Tag(MediumTests.TAG) 051@Tag(ClientTests.TAG) 052public class TestTimestampsFilter { 053 054 private static final Logger LOG = LoggerFactory.getLogger(TestTimestampsFilter.class); 055 056 private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 057 058 private String methodName; 059 060 @BeforeEach 061 public void setUpMethodName(TestInfo testInfo) { 062 this.methodName = testInfo.getTestMethod().get().getName(); 063 } 064 065 @BeforeAll 066 public static void setUpBeforeClass() throws Exception { 067 TEST_UTIL.startMiniCluster(); 068 } 069 070 @AfterAll 071 public static void tearDownAfterClass() throws Exception { 072 TEST_UTIL.shutdownMiniCluster(); 073 } 074 075 /** 076 * @throws java.lang.Exception 077 */ 078 @BeforeEach 079 public void setUp() throws Exception { 080 // Nothing to do. 081 } 082 083 /** 084 * @throws java.lang.Exception 085 */ 086 @AfterEach 087 public void tearDown() throws Exception { 088 // Nothing to do. 089 } 090 091 /** 092 * Test from client side for TimestampsFilter. The TimestampsFilter provides the ability to 093 * request cells (KeyValues) whose timestamp/version is in the specified list of 094 * timestamps/version. 095 */ 096 @Test 097 public void testTimestampsFilter() throws Exception { 098 final byte[] TABLE = Bytes.toBytes(methodName); 099 byte[] FAMILY = Bytes.toBytes("event_log"); 100 byte[][] FAMILIES = new byte[][] { FAMILY }; 101 Cell kvs[]; 102 103 // create table; set versions to max... 104 Table ht = TEST_UTIL.createTable(TableName.valueOf(TABLE), FAMILIES, Integer.MAX_VALUE); 105 106 for (int rowIdx = 0; rowIdx < 5; rowIdx++) { 107 for (int colIdx = 0; colIdx < 5; colIdx++) { 108 // insert versions 201..300 109 putNVersions(ht, FAMILY, rowIdx, colIdx, 201, 300); 110 // insert versions 1..100 111 putNVersions(ht, FAMILY, rowIdx, colIdx, 1, 100); 112 } 113 } 114 115 // do some verification before flush 116 verifyInsertedValues(ht, FAMILY); 117 118 TEST_UTIL.flush(); 119 120 // do some verification after flush 121 verifyInsertedValues(ht, FAMILY); 122 123 // Insert some more versions after flush. These should be in memstore. 124 // After this we should have data in both memstore & HFiles. 125 for (int rowIdx = 0; rowIdx < 5; rowIdx++) { 126 for (int colIdx = 0; colIdx < 5; colIdx++) { 127 putNVersions(ht, FAMILY, rowIdx, colIdx, 301, 400); 128 putNVersions(ht, FAMILY, rowIdx, colIdx, 101, 200); 129 } 130 } 131 132 for (int rowIdx = 0; rowIdx < 5; rowIdx++) { 133 for (int colIdx = 0; colIdx < 5; colIdx++) { 134 kvs = getNVersions(ht, FAMILY, rowIdx, colIdx, Arrays.asList(505L, 5L, 105L, 305L, 205L)); 135 assertEquals(4, kvs.length); 136 checkOneCell(kvs[0], FAMILY, rowIdx, colIdx, 305); 137 checkOneCell(kvs[1], FAMILY, rowIdx, colIdx, 205); 138 checkOneCell(kvs[2], FAMILY, rowIdx, colIdx, 105); 139 checkOneCell(kvs[3], FAMILY, rowIdx, colIdx, 5); 140 } 141 } 142 143 // Request an empty list of versions using the Timestamps filter; 144 // Should return none. 145 kvs = getNVersions(ht, FAMILY, 2, 2, new ArrayList<>()); 146 assertEquals(0, kvs == null ? 0 : kvs.length); 147 148 // 149 // Test the filter using a Scan operation 150 // Scan rows 0..4. For each row, get all its columns, but only 151 // those versions of the columns with the specified timestamps. 152 Result[] results = scanNVersions(ht, FAMILY, 0, 4, Arrays.asList(6L, 106L, 306L)); 153 assertEquals(5, results.length, "# of rows returned from scan"); 154 for (int rowIdx = 0; rowIdx < 5; rowIdx++) { 155 kvs = results[rowIdx].rawCells(); 156 // each row should have 5 columns. 157 // And we have requested 3 versions for each. 158 assertEquals(3 * 5, kvs.length, "Number of KeyValues in result for row:" + rowIdx); 159 for (int colIdx = 0; colIdx < 5; colIdx++) { 160 int offset = colIdx * 3; 161 checkOneCell(kvs[offset + 0], FAMILY, rowIdx, colIdx, 306); 162 checkOneCell(kvs[offset + 1], FAMILY, rowIdx, colIdx, 106); 163 checkOneCell(kvs[offset + 2], FAMILY, rowIdx, colIdx, 6); 164 } 165 } 166 ht.close(); 167 } 168 169 @Test 170 public void testMultiColumns() throws Exception { 171 final byte[] TABLE = Bytes.toBytes(methodName); 172 byte[] FAMILY = Bytes.toBytes("event_log"); 173 byte[][] FAMILIES = new byte[][] { FAMILY }; 174 175 // create table; set versions to max... 176 try (Table ht = TEST_UTIL.createTable(TableName.valueOf(TABLE), FAMILIES, Integer.MAX_VALUE)) { 177 178 Put p = new Put(Bytes.toBytes("row")); 179 p.addColumn(FAMILY, Bytes.toBytes("column0"), 3L, Bytes.toBytes("value0-3")); 180 p.addColumn(FAMILY, Bytes.toBytes("column1"), 3L, Bytes.toBytes("value1-3")); 181 p.addColumn(FAMILY, Bytes.toBytes("column2"), 1L, Bytes.toBytes("value2-1")); 182 p.addColumn(FAMILY, Bytes.toBytes("column2"), 2L, Bytes.toBytes("value2-2")); 183 p.addColumn(FAMILY, Bytes.toBytes("column2"), 3L, Bytes.toBytes("value2-3")); 184 p.addColumn(FAMILY, Bytes.toBytes("column3"), 2L, Bytes.toBytes("value3-2")); 185 p.addColumn(FAMILY, Bytes.toBytes("column4"), 1L, Bytes.toBytes("value4-1")); 186 p.addColumn(FAMILY, Bytes.toBytes("column4"), 2L, Bytes.toBytes("value4-2")); 187 p.addColumn(FAMILY, Bytes.toBytes("column4"), 3L, Bytes.toBytes("value4-3")); 188 ht.put(p); 189 190 ArrayList<Long> timestamps = new ArrayList<>(); 191 timestamps.add(3L); 192 TimestampsFilter filter = new TimestampsFilter(timestamps); 193 194 Get g = new Get(Bytes.toBytes("row")); 195 g.setFilter(filter); 196 g.readAllVersions(); 197 g.addColumn(FAMILY, Bytes.toBytes("column2")); 198 g.addColumn(FAMILY, Bytes.toBytes("column4")); 199 200 Result result = ht.get(g); 201 for (Cell kv : result.listCells()) { 202 LOG.info("found row " + Bytes.toString(CellUtil.cloneRow(kv)) + ", column " 203 + Bytes.toString(CellUtil.cloneQualifier(kv)) + ", value " 204 + Bytes.toString(CellUtil.cloneValue(kv))); 205 } 206 207 assertEquals(2, result.listCells().size()); 208 assertTrue(CellUtil.matchingValue(result.listCells().get(0), Bytes.toBytes("value2-3"))); 209 assertTrue(CellUtil.matchingValue(result.listCells().get(1), Bytes.toBytes("value4-3"))); 210 } 211 } 212 213 /** 214 * Test TimestampsFilter in the presence of version deletes. 215 */ 216 @Test 217 public void testWithVersionDeletes() throws Exception { 218 219 // first test from memstore (without flushing). 220 testWithVersionDeletes(false); 221 222 // run same test against HFiles (by forcing a flush). 223 testWithVersionDeletes(true); 224 } 225 226 private void testWithVersionDeletes(boolean flushTables) throws IOException { 227 final byte[] TABLE = Bytes.toBytes(methodName + "_" + (flushTables ? "flush" : "noflush")); 228 byte[] FAMILY = Bytes.toBytes("event_log"); 229 byte[][] FAMILIES = new byte[][] { FAMILY }; 230 231 // create table; set versions to max... 232 Table ht = TEST_UTIL.createTable(TableName.valueOf(TABLE), FAMILIES, Integer.MAX_VALUE); 233 234 // For row:0, col:0: insert versions 1 through 5. 235 putNVersions(ht, FAMILY, 0, 0, 1, 5); 236 237 // delete version 4. 238 deleteOneVersion(ht, FAMILY, 0, 0, 4); 239 240 if (flushTables) { 241 TEST_UTIL.flush(); 242 } 243 244 // request a bunch of versions including the deleted version. We should 245 // only get back entries for the versions that exist. 246 Cell kvs[] = getNVersions(ht, FAMILY, 0, 0, Arrays.asList(2L, 3L, 4L, 5L)); 247 assertEquals(3, kvs.length); 248 checkOneCell(kvs[0], FAMILY, 0, 0, 5); 249 checkOneCell(kvs[1], FAMILY, 0, 0, 3); 250 checkOneCell(kvs[2], FAMILY, 0, 0, 2); 251 252 ht.close(); 253 } 254 255 private void verifyInsertedValues(Table ht, byte[] cf) throws IOException { 256 for (int rowIdx = 0; rowIdx < 5; rowIdx++) { 257 for (int colIdx = 0; colIdx < 5; colIdx++) { 258 // ask for versions that exist. 259 Cell[] kvs = getNVersions(ht, cf, rowIdx, colIdx, Arrays.asList(5L, 300L, 6L, 80L)); 260 assertEquals(4, kvs.length); 261 checkOneCell(kvs[0], cf, rowIdx, colIdx, 300); 262 checkOneCell(kvs[1], cf, rowIdx, colIdx, 80); 263 checkOneCell(kvs[2], cf, rowIdx, colIdx, 6); 264 checkOneCell(kvs[3], cf, rowIdx, colIdx, 5); 265 266 // ask for versions that do not exist. 267 kvs = getNVersions(ht, cf, rowIdx, colIdx, Arrays.asList(101L, 102L)); 268 assertEquals(0, kvs == null ? 0 : kvs.length); 269 270 // ask for some versions that exist and some that do not. 271 kvs = getNVersions(ht, cf, rowIdx, colIdx, Arrays.asList(1L, 300L, 105L, 70L, 115L)); 272 assertEquals(3, kvs.length); 273 checkOneCell(kvs[0], cf, rowIdx, colIdx, 300); 274 checkOneCell(kvs[1], cf, rowIdx, colIdx, 70); 275 checkOneCell(kvs[2], cf, rowIdx, colIdx, 1); 276 } 277 } 278 } 279 280 /** 281 * Assert that the passed in KeyValue has expected contents for the specified row, column & 282 * timestamp. 283 */ 284 private void checkOneCell(Cell kv, byte[] cf, int rowIdx, int colIdx, long ts) { 285 286 String ctx = "rowIdx=" + rowIdx + "; colIdx=" + colIdx + "; ts=" + ts; 287 288 assertEquals("row:" + rowIdx, Bytes.toString(CellUtil.cloneRow(kv)), 289 "Row mismatch which checking: " + ctx); 290 291 assertEquals(Bytes.toString(cf), Bytes.toString(CellUtil.cloneFamily(kv)), 292 "ColumnFamily mismatch while checking: " + ctx); 293 294 assertEquals("column:" + colIdx, Bytes.toString(CellUtil.cloneQualifier(kv)), 295 "Column qualifier mismatch while checking: " + ctx); 296 297 assertEquals(ts, kv.getTimestamp(), "Timestamp mismatch while checking: " + ctx); 298 299 assertEquals("value-version-" + ts, Bytes.toString(CellUtil.cloneValue(kv)), 300 "Value mismatch while checking: " + ctx); 301 } 302 303 /** 304 * Uses the TimestampFilter on a Get to request a specified list of versions for the row/column 305 * specified by rowIdx & colIdx. 306 */ 307 private Cell[] getNVersions(Table ht, byte[] cf, int rowIdx, int colIdx, List<Long> versions) 308 throws IOException { 309 byte row[] = Bytes.toBytes("row:" + rowIdx); 310 byte column[] = Bytes.toBytes("column:" + colIdx); 311 Filter filter = new TimestampsFilter(versions); 312 Get get = new Get(row); 313 get.addColumn(cf, column); 314 get.setFilter(filter); 315 get.readAllVersions(); 316 Result result = ht.get(get); 317 318 return result.rawCells(); 319 } 320 321 /** 322 * Uses the TimestampFilter on a Scan to request a specified list of versions for the rows from 323 * startRowIdx to endRowIdx (both inclusive). 324 */ 325 private Result[] scanNVersions(Table ht, byte[] cf, int startRowIdx, int endRowIdx, 326 List<Long> versions) throws IOException { 327 byte startRow[] = Bytes.toBytes("row:" + startRowIdx); 328 byte endRow[] = Bytes.toBytes("row:" + endRowIdx + 1); // exclusive 329 Filter filter = new TimestampsFilter(versions); 330 Scan scan = new Scan().withStartRow(startRow).withStopRow(endRow); 331 scan.setFilter(filter); 332 scan.readAllVersions(); 333 ResultScanner scanner = ht.getScanner(scan); 334 return scanner.next(endRowIdx - startRowIdx + 1); 335 } 336 337 /** 338 * Insert in specific row/column versions with timestamps versionStart..versionEnd. 339 */ 340 private void putNVersions(Table ht, byte[] cf, int rowIdx, int colIdx, long versionStart, 341 long versionEnd) throws IOException { 342 byte row[] = Bytes.toBytes("row:" + rowIdx); 343 byte column[] = Bytes.toBytes("column:" + colIdx); 344 Put put = new Put(row); 345 put.setDurability(Durability.SKIP_WAL); 346 347 for (long idx = versionStart; idx <= versionEnd; idx++) { 348 put.addColumn(cf, column, idx, Bytes.toBytes("value-version-" + idx)); 349 } 350 351 ht.put(put); 352 } 353 354 /** 355 * For row/column specified by rowIdx/colIdx, delete the cell corresponding to the specified 356 * version. 357 */ 358 private void deleteOneVersion(Table ht, byte[] cf, int rowIdx, int colIdx, long version) 359 throws IOException { 360 byte row[] = Bytes.toBytes("row:" + rowIdx); 361 byte column[] = Bytes.toBytes("column:" + colIdx); 362 Delete del = new Delete(row); 363 del.addColumn(cf, column, version); 364 ht.delete(del); 365 } 366 367}