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.io.compress.zstd;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertNotNull;
022import static org.junit.jupiter.api.Assertions.assertTrue;
023
024import java.util.concurrent.TimeUnit;
025import org.apache.hadoop.conf.Configuration;
026import org.apache.hadoop.hbase.HBaseTestingUtil;
027import org.apache.hadoop.hbase.HConstants;
028import org.apache.hadoop.hbase.ServerName;
029import org.apache.hadoop.hbase.TableName;
030import org.apache.hadoop.hbase.Waiter.ExplainingPredicate;
031import org.apache.hadoop.hbase.client.Admin;
032import org.apache.hadoop.hbase.client.AsyncConnection;
033import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
034import org.apache.hadoop.hbase.client.ConnectionFactory;
035import org.apache.hadoop.hbase.client.RegionInfo;
036import org.apache.hadoop.hbase.client.Table;
037import org.apache.hadoop.hbase.client.TableDescriptor;
038import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
039import org.apache.hadoop.hbase.io.compress.Compression;
040import org.apache.hadoop.hbase.io.compress.DictionaryCache;
041import org.apache.hadoop.hbase.testclassification.LargeTests;
042import org.apache.hadoop.hbase.testclassification.RegionServerTests;
043import org.apache.hadoop.hbase.util.Bytes;
044import org.junit.jupiter.api.AfterAll;
045import org.junit.jupiter.api.BeforeAll;
046import org.junit.jupiter.api.Tag;
047import org.junit.jupiter.api.Test;
048
049@Tag(RegionServerTests.TAG)
050@Tag(LargeTests.TAG)
051public class TestZstdDictionarySplitMerge {
052
053  private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
054  private static int numRows = 10_000;
055  private static Configuration conf;
056
057  @BeforeAll
058  public static void setUp() throws Exception {
059    // NOTE: Don't put configuration settings in global site schema. We are testing if per
060    // CF or per table schema settings are applied correctly.
061    conf = TEST_UTIL.getConfiguration();
062    conf.set(Compression.ZSTD_CODEC_CLASS_KEY, ZstdCodec.class.getCanonicalName());
063    Compression.Algorithm.ZSTD.reload(conf);
064    conf.setInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, 1000);
065    conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2);
066    TEST_UTIL.startMiniCluster(1);
067  }
068
069  @AfterAll
070  public static void tearDown() throws Exception {
071    TEST_UTIL.shutdownMiniCluster();
072  }
073
074  @Test
075  public void test() throws Exception {
076    // Create the table
077    final TableName tableName = TableName.valueOf("TestZstdDictionarySplitMerge");
078    final byte[] cfName = Bytes.toBytes("info");
079    final String dictionaryPath = DictionaryCache.RESOURCE_SCHEME + "zstd.test.dict";
080    final TableDescriptor td = TableDescriptorBuilder.newBuilder(tableName)
081      .setColumnFamily(ColumnFamilyDescriptorBuilder.newBuilder(cfName)
082        .setCompressionType(Compression.Algorithm.ZSTD)
083        .setConfiguration(ZstdCodec.ZSTD_DICTIONARY_KEY, dictionaryPath).build())
084      .build();
085    final Admin admin = TEST_UTIL.getAdmin();
086    admin.createTable(td, new byte[][] { Bytes.toBytes(String.valueOf(1)) });
087    TEST_UTIL.waitTableAvailable(tableName);
088    // Load some data
089    Table t = ConnectionFactory.createConnection(conf).getTable(tableName);
090    TEST_UTIL.loadNumericRows(t, cfName, 0, numRows);
091    admin.flush(tableName);
092    assertTrue(DictionaryCache.contains(dictionaryPath), "Dictionary was not loaded");
093    TEST_UTIL.verifyNumericRows(t, cfName, 0, numRows, 0);
094    // Test split procedure
095    admin.split(tableName, Bytes.toBytes(String.valueOf(numRows / 2)));
096    TEST_UTIL.waitFor(30000, new ExplainingPredicate<Exception>() {
097      @Override
098      public boolean evaluate() throws Exception {
099        return TEST_UTIL.getMiniHBaseCluster().getRegions(tableName).size() == 3;
100      }
101
102      @Override
103      public String explainFailure() throws Exception {
104        return "Split has not finished yet";
105      }
106    });
107    TEST_UTIL.waitUntilNoRegionsInTransition();
108    TEST_UTIL.verifyNumericRows(t, cfName, 0, numRows, 0);
109    // Test merge procedure
110    RegionInfo regionA = null;
111    RegionInfo regionB = null;
112
113    for (RegionInfo region : admin.getRegions(tableName)) {
114      if (region.getStartKey().length == 0) {
115        regionA = region;
116      } else if (Bytes.equals(region.getStartKey(), Bytes.toBytes(String.valueOf(1)))) {
117        regionB = region;
118      }
119    }
120    assertNotNull(regionA);
121    assertNotNull(regionB);
122
123    // major compact before merging otherwise the regions are not mergable
124    TEST_UTIL.compact(tableName, true);
125    admin.reopenTableRegions(tableName);
126
127    admin
128      .mergeRegionsAsync(new byte[][] { regionA.getRegionName(), regionB.getRegionName() }, false)
129      .get(300, TimeUnit.SECONDS);
130    assertEquals(2, admin.getRegions(tableName).size());
131    ServerName expected = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0).getServerName();
132    assertEquals(expected, TEST_UTIL.getConnection().getRegionLocator(tableName)
133      .getRegionLocation(Bytes.toBytes(String.valueOf(1)), true).getServerName());
134    try (AsyncConnection asyncConn = ConnectionFactory.createAsyncConnection(conf).get()) {
135      assertEquals(expected, asyncConn.getRegionLocator(tableName)
136        .getRegionLocation(Bytes.toBytes(String.valueOf(1)), true).get().getServerName());
137    }
138    TEST_UTIL.verifyNumericRows(t, cfName, 0, numRows, 0);
139  }
140}