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}