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.assertFalse;
022import static org.junit.jupiter.api.Assertions.assertTrue;
023
024import java.util.List;
025import java.util.concurrent.ExecutionException;
026import java.util.stream.Collectors;
027import java.util.stream.IntStream;
028import org.apache.hadoop.hbase.HBaseTestingUtil;
029import org.apache.hadoop.hbase.TableName;
030import org.apache.hadoop.hbase.testclassification.ClientTests;
031import org.apache.hadoop.hbase.testclassification.MediumTests;
032import org.apache.hadoop.hbase.util.Bytes;
033import org.junit.jupiter.api.AfterAll;
034import org.junit.jupiter.api.BeforeAll;
035import org.junit.jupiter.api.Tag;
036import org.junit.jupiter.api.Test;
037
038/**
039 * With filter we may stop at a middle of row and think that we still have more cells for the
040 * current row but actually all the remaining cells will be filtered out by the filter. So it will
041 * lead to a Result that mayHaveMoreCellsInRow is true but actually there are no cells for the same
042 * row. Here we want to test if our limited scan still works.
043 */
044@Tag(MediumTests.TAG)
045@Tag(ClientTests.TAG)
046public class TestRawAsyncTableLimitedScanWithFilter {
047
048  private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
049
050  private static final TableName TABLE_NAME = TableName.valueOf("TestRegionScanner");
051
052  private static final byte[] FAMILY = Bytes.toBytes("cf");
053
054  private static final byte[][] CQS =
055    { Bytes.toBytes("cq1"), Bytes.toBytes("cq2"), Bytes.toBytes("cq3"), Bytes.toBytes("cq4") };
056
057  private static int ROW_COUNT = 10;
058
059  private static AsyncConnection CONN;
060
061  private static AsyncTable<?> TABLE;
062
063  @BeforeAll
064  public static void setUp() throws Exception {
065    UTIL.startMiniCluster(1);
066    UTIL.createTable(TABLE_NAME, FAMILY);
067    CONN = ConnectionFactory.createAsyncConnection(UTIL.getConfiguration()).get();
068    TABLE = CONN.getTable(TABLE_NAME);
069    TABLE.putAll(IntStream.range(0, ROW_COUNT).mapToObj(i -> {
070      Put put = new Put(Bytes.toBytes(i));
071      IntStream.range(0, CQS.length)
072        .forEach(j -> put.addColumn(FAMILY, CQS[j], Bytes.toBytes((j + 1) * i)));
073      return put;
074    }).collect(Collectors.toList())).get();
075  }
076
077  @AfterAll
078  public static void tearDown() throws Exception {
079    if (CONN != null) {
080      CONN.close();
081    }
082    UTIL.shutdownMiniCluster();
083  }
084
085  @Test
086  public void testCompleteResult() throws InterruptedException, ExecutionException {
087    int limit = 5;
088    Scan scan =
089      new Scan().setFilter(new ColumnCountOnRowFilter(2)).setMaxResultSize(1).setLimit(limit);
090    List<Result> results = TABLE.scanAll(scan).get();
091    assertEquals(limit, results.size());
092    IntStream.range(0, limit).forEach(i -> {
093      Result result = results.get(i);
094      assertEquals(i, Bytes.toInt(result.getRow()));
095      assertEquals(2, result.size());
096      assertFalse(result.mayHaveMoreCellsInRow());
097      assertEquals(i, Bytes.toInt(result.getValue(FAMILY, CQS[0])));
098      assertEquals(2 * i, Bytes.toInt(result.getValue(FAMILY, CQS[1])));
099    });
100  }
101
102  @Test
103  public void testAllowPartial() throws InterruptedException, ExecutionException {
104    int limit = 5;
105    Scan scan = new Scan().setFilter(new ColumnCountOnRowFilter(2)).setMaxResultSize(1)
106      .setAllowPartialResults(true).setLimit(limit);
107    List<Result> results = TABLE.scanAll(scan).get();
108    assertEquals(2 * limit, results.size());
109    IntStream.range(0, 2 * limit).forEach(i -> {
110      int key = i / 2;
111      Result result = results.get(i);
112      assertEquals(key, Bytes.toInt(result.getRow()));
113      assertEquals(1, result.size());
114      assertTrue(result.mayHaveMoreCellsInRow());
115      int cqIndex = i % 2;
116      assertEquals(key * (cqIndex + 1), Bytes.toInt(result.getValue(FAMILY, CQS[cqIndex])));
117    });
118  }
119
120  @Test
121  public void testBatchAllowPartial() throws InterruptedException, ExecutionException {
122    int limit = 5;
123    Scan scan = new Scan().setFilter(new ColumnCountOnRowFilter(3)).setBatch(2).setMaxResultSize(1)
124      .setAllowPartialResults(true).setLimit(limit);
125    List<Result> results = TABLE.scanAll(scan).get();
126    assertEquals(3 * limit, results.size());
127    IntStream.range(0, 3 * limit).forEach(i -> {
128      int key = i / 3;
129      Result result = results.get(i);
130      assertEquals(key, Bytes.toInt(result.getRow()));
131      assertEquals(1, result.size());
132      assertTrue(result.mayHaveMoreCellsInRow());
133      int cqIndex = i % 3;
134      assertEquals(key * (cqIndex + 1), Bytes.toInt(result.getValue(FAMILY, CQS[cqIndex])));
135    });
136  }
137
138  @Test
139  public void testBatch() throws InterruptedException, ExecutionException {
140    int limit = 5;
141    Scan scan = new Scan().setFilter(new ColumnCountOnRowFilter(2)).setBatch(2).setMaxResultSize(1)
142      .setLimit(limit);
143    List<Result> results = TABLE.scanAll(scan).get();
144    assertEquals(limit, results.size());
145    IntStream.range(0, limit).forEach(i -> {
146      Result result = results.get(i);
147      assertEquals(i, Bytes.toInt(result.getRow()));
148      assertEquals(2, result.size());
149      assertTrue(result.mayHaveMoreCellsInRow());
150      assertEquals(i, Bytes.toInt(result.getValue(FAMILY, CQS[0])));
151      assertEquals(2 * i, Bytes.toInt(result.getValue(FAMILY, CQS[1])));
152    });
153  }
154
155  @Test
156  public void testBatchAndFilterDiffer() throws InterruptedException, ExecutionException {
157    int limit = 5;
158    Scan scan = new Scan().setFilter(new ColumnCountOnRowFilter(3)).setBatch(2).setMaxResultSize(1)
159      .setLimit(limit);
160    List<Result> results = TABLE.scanAll(scan).get();
161    assertEquals(2 * limit, results.size());
162    IntStream.range(0, limit).forEach(i -> {
163      Result result = results.get(2 * i);
164      assertEquals(i, Bytes.toInt(result.getRow()));
165      assertEquals(2, result.size());
166      assertTrue(result.mayHaveMoreCellsInRow());
167      assertEquals(i, Bytes.toInt(result.getValue(FAMILY, CQS[0])));
168      assertEquals(2 * i, Bytes.toInt(result.getValue(FAMILY, CQS[1])));
169      result = results.get(2 * i + 1);
170      assertEquals(i, Bytes.toInt(result.getRow()));
171      assertEquals(1, result.size());
172      assertFalse(result.mayHaveMoreCellsInRow());
173      assertEquals(3 * i, Bytes.toInt(result.getValue(FAMILY, CQS[2])));
174    });
175  }
176}