You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2017/09/03 12:04:43 UTC
[2/4] hbase git commit: HBASE-18699 Copy LoadIncrementalHFiles to
another package and mark the old one as deprecated
http://git-wip-us.apache.org/repos/asf/hbase/blob/9e53f292/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java
new file mode 100644
index 0000000..1f27d04
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/tool/LoadIncrementalHFiles.java
@@ -0,0 +1,1251 @@
+/**
+ * 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.hadoop.hbase.tool;
+
+import static java.lang.String.format;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.SortedMap;
+import java.util.TreeMap;
+import java.util.UUID;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import org.apache.commons.lang.mutable.MutableInt;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.TableNotFoundException;
+import org.apache.hadoop.hbase.classification.InterfaceAudience;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.ClientServiceCallable;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
+import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.RegionLocator;
+import org.apache.hadoop.hbase.client.RpcRetryingCallerFactory;
+import org.apache.hadoop.hbase.client.SecureBulkLoadClient;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
+import org.apache.hadoop.hbase.io.HFileLink;
+import org.apache.hadoop.hbase.io.HalfStoreFileReader;
+import org.apache.hadoop.hbase.io.Reference;
+import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
+import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.io.hfile.HFile;
+import org.apache.hadoop.hbase.io.hfile.HFileContext;
+import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
+import org.apache.hadoop.hbase.io.hfile.HFileDataBlockEncoder;
+import org.apache.hadoop.hbase.io.hfile.HFileScanner;
+import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
+import org.apache.hadoop.hbase.regionserver.BloomType;
+import org.apache.hadoop.hbase.regionserver.HStore;
+import org.apache.hadoop.hbase.regionserver.StoreFileInfo;
+import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
+import org.apache.hadoop.hbase.security.UserProvider;
+import org.apache.hadoop.hbase.security.token.FsDelegationToken;
+import org.apache.hadoop.hbase.shaded.com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hbase.shaded.com.google.common.collect.HashMultimap;
+import org.apache.hadoop.hbase.shaded.com.google.common.collect.Multimap;
+import org.apache.hadoop.hbase.shaded.com.google.common.collect.Multimaps;
+import org.apache.hadoop.hbase.shaded.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.FSHDFSUtils;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+
+/**
+ * Tool to load the output of HFileOutputFormat into an existing table.
+ */
+@InterfaceAudience.Public
+public class LoadIncrementalHFiles extends Configured implements Tool {
+
+ private static final Log LOG = LogFactory.getLog(LoadIncrementalHFiles.class);
+
+ public static final String NAME = "completebulkload";
+ static final String RETRY_ON_IO_EXCEPTION = "hbase.bulkload.retries.retryOnIOException";
+ public static final String MAX_FILES_PER_REGION_PER_FAMILY =
+ "hbase.mapreduce.bulkload.max.hfiles.perRegion.perFamily";
+ private static final String ASSIGN_SEQ_IDS = "hbase.mapreduce.bulkload.assign.sequenceNumbers";
+ public final static String CREATE_TABLE_CONF_KEY = "create.table";
+ public final static String IGNORE_UNMATCHED_CF_CONF_KEY = "ignore.unmatched.families";
+ public final static String ALWAYS_COPY_FILES = "always.copy.files";
+
+ // We use a '.' prefix which is ignored when walking directory trees
+ // above. It is invalid family name.
+ static final String TMP_DIR = ".tmp";
+
+ private final int maxFilesPerRegionPerFamily;
+ private final boolean assignSeqIds;
+
+ // Source delegation token
+ private final FsDelegationToken fsDelegationToken;
+ private final UserProvider userProvider;
+ private final int nrThreads;
+ private final RpcControllerFactory rpcControllerFactory;
+
+ private String bulkToken;
+
+ /**
+ * Represents an HFile waiting to be loaded. An queue is used in this class in order to support
+ * the case where a region has split during the process of the load. When this happens, the HFile
+ * is split into two physical parts across the new region boundary, and each part is added back
+ * into the queue. The import process finishes when the queue is empty.
+ */
+ @InterfaceAudience.Public
+ public static class LoadQueueItem {
+ private final byte[] family;
+ private final Path hfilePath;
+
+ public LoadQueueItem(byte[] family, Path hfilePath) {
+ this.family = family;
+ this.hfilePath = hfilePath;
+ }
+
+ @Override
+ public String toString() {
+ return "family:" + Bytes.toString(family) + " path:" + hfilePath.toString();
+ }
+
+ public byte[] getFamily() {
+ return family;
+ }
+
+ public Path getFilePath() {
+ return hfilePath;
+ }
+ }
+
+ public LoadIncrementalHFiles(Configuration conf) {
+ // make a copy, just to be sure we're not overriding someone else's config
+ super(HBaseConfiguration.create(conf));
+ conf = getConf();
+ // disable blockcache for tool invocation, see HBASE-10500
+ conf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0);
+ userProvider = UserProvider.instantiate(conf);
+ fsDelegationToken = new FsDelegationToken(userProvider, "renewer");
+ assignSeqIds = conf.getBoolean(ASSIGN_SEQ_IDS, true);
+ maxFilesPerRegionPerFamily = conf.getInt(MAX_FILES_PER_REGION_PER_FAMILY, 32);
+ nrThreads = conf.getInt("hbase.loadincremental.threads.max",
+ Runtime.getRuntime().availableProcessors());
+ rpcControllerFactory = new RpcControllerFactory(conf);
+ }
+
+ private void usage() {
+ System.err.println("usage: " + NAME + " /path/to/hfileoutputformat-output tablename" + "\n -D" +
+ CREATE_TABLE_CONF_KEY + "=no - can be used to avoid creation of table by this tool\n" +
+ " Note: if you set this to 'no', then the target table must already exist in HBase\n -D" +
+ IGNORE_UNMATCHED_CF_CONF_KEY + "=yes - can be used to ignore unmatched column families\n" +
+ "\n");
+ }
+
+ /**
+ * Prepare a collection of {@link LoadQueueItem} from list of source hfiles contained in the
+ * passed directory and validates whether the prepared queue has all the valid table column
+ * families in it.
+ * @param hfilesDir directory containing list of hfiles to be loaded into the table
+ * @param table table to which hfiles should be loaded
+ * @param queue queue which needs to be loaded into the table
+ * @param validateHFile if true hfiles will be validated for its format
+ * @throws IOException If any I/O or network error occurred
+ */
+ public void prepareHFileQueue(Path hfilesDir, Table table, Deque<LoadQueueItem> queue,
+ boolean validateHFile) throws IOException {
+ prepareHFileQueue(hfilesDir, table, queue, validateHFile, false);
+ }
+
+ /**
+ * Prepare a collection of {@link LoadQueueItem} from list of source hfiles contained in the
+ * passed directory and validates whether the prepared queue has all the valid table column
+ * families in it.
+ * @param hfilesDir directory containing list of hfiles to be loaded into the table
+ * @param table table to which hfiles should be loaded
+ * @param queue queue which needs to be loaded into the table
+ * @param validateHFile if true hfiles will be validated for its format
+ * @param silence true to ignore unmatched column families
+ * @throws IOException If any I/O or network error occurred
+ */
+ public void prepareHFileQueue(Path hfilesDir, Table table, Deque<LoadQueueItem> queue,
+ boolean validateHFile, boolean silence) throws IOException {
+ discoverLoadQueue(queue, hfilesDir, validateHFile);
+ validateFamiliesInHFiles(table, queue, silence);
+ }
+
+ /**
+ * Prepare a collection of {@link LoadQueueItem} from list of source hfiles contained in the
+ * passed directory and validates whether the prepared queue has all the valid table column
+ * families in it.
+ * @param map map of family to List of hfiles
+ * @param table table to which hfiles should be loaded
+ * @param queue queue which needs to be loaded into the table
+ * @param silence true to ignore unmatched column families
+ * @throws IOException If any I/O or network error occurred
+ */
+ public void prepareHFileQueue(Map<byte[], List<Path>> map, Table table,
+ Deque<LoadQueueItem> queue, boolean silence) throws IOException {
+ populateLoadQueue(queue, map);
+ validateFamiliesInHFiles(table, queue, silence);
+ }
+
+ /**
+ * Perform a bulk load of the given directory into the given pre-existing table. This method is
+ * not threadsafe.
+ * @param hfofDir the directory that was provided as the output path of a job using
+ * HFileOutputFormat
+ * @param admin the Admin
+ * @param table the table to load into
+ * @param regionLocator region locator
+ * @throws TableNotFoundException if table does not yet exist
+ */
+ public Map<LoadQueueItem, ByteBuffer> doBulkLoad(Path hfofDir, final Admin admin, Table table,
+ RegionLocator regionLocator) throws TableNotFoundException, IOException {
+ return doBulkLoad(hfofDir, admin, table, regionLocator, false, false);
+ }
+
+ /**
+ * Perform a bulk load of the given directory into the given pre-existing table. This method is
+ * not threadsafe.
+ * @param map map of family to List of hfiles
+ * @param admin the Admin
+ * @param table the table to load into
+ * @param regionLocator region locator
+ * @param silence true to ignore unmatched column families
+ * @param copyFile always copy hfiles if true
+ * @throws TableNotFoundException if table does not yet exist
+ */
+ public Map<LoadQueueItem, ByteBuffer> doBulkLoad(Map<byte[], List<Path>> map, final Admin admin,
+ Table table, RegionLocator regionLocator, boolean silence, boolean copyFile)
+ throws TableNotFoundException, IOException {
+ if (!admin.isTableAvailable(regionLocator.getName())) {
+ throw new TableNotFoundException("Table " + table.getName() + " is not currently available.");
+ }
+ // LQI queue does not need to be threadsafe -- all operations on this queue
+ // happen in this thread
+ Deque<LoadQueueItem> queue = new ArrayDeque<>();
+ ExecutorService pool = null;
+ SecureBulkLoadClient secureClient = null;
+ try {
+ prepareHFileQueue(map, table, queue, silence);
+ if (queue.isEmpty()) {
+ LOG.warn("Bulk load operation did not get any files to load");
+ return Collections.emptyMap();
+ }
+ pool = createExecutorService();
+ secureClient = new SecureBulkLoadClient(table.getConfiguration(), table);
+ return performBulkLoad(admin, table, regionLocator, queue, pool, secureClient, copyFile);
+ } finally {
+ cleanup(admin, queue, pool, secureClient);
+ }
+ }
+
+ /**
+ * Perform a bulk load of the given directory into the given pre-existing table. This method is
+ * not threadsafe.
+ * @param hfofDir the directory that was provided as the output path of a job using
+ * HFileOutputFormat
+ * @param admin the Admin
+ * @param table the table to load into
+ * @param regionLocator region locator
+ * @param silence true to ignore unmatched column families
+ * @param copyFile always copy hfiles if true
+ * @throws TableNotFoundException if table does not yet exist
+ */
+ public Map<LoadQueueItem, ByteBuffer> doBulkLoad(Path hfofDir, final Admin admin, Table table,
+ RegionLocator regionLocator, boolean silence, boolean copyFile)
+ throws TableNotFoundException, IOException {
+ if (!admin.isTableAvailable(regionLocator.getName())) {
+ throw new TableNotFoundException("Table " + table.getName() + " is not currently available.");
+ }
+
+ /*
+ * Checking hfile format is a time-consuming operation, we should have an option to skip this
+ * step when bulkloading millions of HFiles. See HBASE-13985.
+ */
+ boolean validateHFile = getConf().getBoolean("hbase.loadincremental.validate.hfile", true);
+ if (!validateHFile) {
+ LOG.warn("You are skipping HFiles validation, it might cause some data loss if files " +
+ "are not correct. If you fail to read data from your table after using this " +
+ "option, consider removing the files and bulkload again without this option. " +
+ "See HBASE-13985");
+ }
+ // LQI queue does not need to be threadsafe -- all operations on this queue
+ // happen in this thread
+ Deque<LoadQueueItem> queue = new ArrayDeque<>();
+ ExecutorService pool = null;
+ SecureBulkLoadClient secureClient = null;
+ try {
+ prepareHFileQueue(hfofDir, table, queue, validateHFile, silence);
+
+ if (queue.isEmpty()) {
+ LOG.warn(
+ "Bulk load operation did not find any files to load in " + "directory " + hfofDir != null
+ ? hfofDir.toUri()
+ : "" + ". Does it contain files in " +
+ "subdirectories that correspond to column family names?");
+ return Collections.emptyMap();
+ }
+ pool = createExecutorService();
+ secureClient = new SecureBulkLoadClient(table.getConfiguration(), table);
+ return performBulkLoad(admin, table, regionLocator, queue, pool, secureClient, copyFile);
+ } finally {
+ cleanup(admin, queue, pool, secureClient);
+ }
+ }
+
+ /**
+ * Used by the replication sink to load the hfiles from the source cluster. It does the following,
+ * <ol>
+ * <li>LoadIncrementalHFiles#groupOrSplitPhase(Table, ExecutorService, Deque, Pair)}</li>
+ * <li>LoadIncrementalHFiles#bulkLoadPhase(Table, Connection, ExecutorService, Deque, Multimap)
+ * </li>
+ * </ol>
+ * @param table Table to which these hfiles should be loaded to
+ * @param conn Connection to use
+ * @param queue {@link LoadQueueItem} has hfiles yet to be loaded
+ * @param startEndKeys starting and ending row keys of the region
+ */
+ public void loadHFileQueue(Table table, Connection conn, Deque<LoadQueueItem> queue,
+ Pair<byte[][], byte[][]> startEndKeys) throws IOException {
+ loadHFileQueue(table, conn, queue, startEndKeys, false);
+ }
+
+ /**
+ * Used by the replication sink to load the hfiles from the source cluster. It does the following,
+ * <ol>
+ * <li>LoadIncrementalHFiles#groupOrSplitPhase(Table, ExecutorService, Deque, Pair)}</li>
+ * <li>LoadIncrementalHFiles#bulkLoadPhase(Table, Connection, ExecutorService, Deque, Multimap)
+ * </li>
+ * </ol>
+ * @param table Table to which these hfiles should be loaded to
+ * @param conn Connection to use
+ * @param queue {@link LoadQueueItem} has hfiles yet to be loaded
+ * @param startEndKeys starting and ending row keys of the region
+ */
+ public void loadHFileQueue(Table table, Connection conn, Deque<LoadQueueItem> queue,
+ Pair<byte[][], byte[][]> startEndKeys, boolean copyFile) throws IOException {
+ ExecutorService pool = null;
+ try {
+ pool = createExecutorService();
+ Multimap<ByteBuffer, LoadQueueItem> regionGroups =
+ groupOrSplitPhase(table, pool, queue, startEndKeys).getFirst();
+ bulkLoadPhase(table, conn, pool, queue, regionGroups, copyFile, null);
+ } finally {
+ if (pool != null) {
+ pool.shutdown();
+ }
+ }
+ }
+
+ private Map<LoadQueueItem, ByteBuffer> performBulkLoad(Admin admin, Table table,
+ RegionLocator regionLocator, Deque<LoadQueueItem> queue, ExecutorService pool,
+ SecureBulkLoadClient secureClient, boolean copyFile) throws IOException {
+ int count = 0;
+
+ if (isSecureBulkLoadEndpointAvailable()) {
+ LOG.warn("SecureBulkLoadEndpoint is deprecated. It will be removed in future releases.");
+ LOG.warn("Secure bulk load has been integrated into HBase core.");
+ }
+
+ fsDelegationToken.acquireDelegationToken(queue.peek().getFilePath().getFileSystem(getConf()));
+ bulkToken = secureClient.prepareBulkLoad(admin.getConnection());
+ Pair<Multimap<ByteBuffer, LoadQueueItem>, Set<String>> pair = null;
+
+ Map<LoadQueueItem, ByteBuffer> item2RegionMap = new HashMap<>();
+ // Assumes that region splits can happen while this occurs.
+ while (!queue.isEmpty()) {
+ // need to reload split keys each iteration.
+ final Pair<byte[][], byte[][]> startEndKeys = regionLocator.getStartEndKeys();
+ if (count != 0) {
+ LOG.info("Split occurred while grouping HFiles, retry attempt " + +count + " with " +
+ queue.size() + " files remaining to group or split");
+ }
+
+ int maxRetries = getConf().getInt(HConstants.BULKLOAD_MAX_RETRIES_NUMBER, 10);
+ maxRetries = Math.max(maxRetries, startEndKeys.getFirst().length + 1);
+ if (maxRetries != 0 && count >= maxRetries) {
+ throw new IOException(
+ "Retry attempted " + count + " times without completing, bailing out");
+ }
+ count++;
+
+ // Using ByteBuffer for byte[] equality semantics
+ pair = groupOrSplitPhase(table, pool, queue, startEndKeys);
+ Multimap<ByteBuffer, LoadQueueItem> regionGroups = pair.getFirst();
+
+ if (!checkHFilesCountPerRegionPerFamily(regionGroups)) {
+ // Error is logged inside checkHFilesCountPerRegionPerFamily.
+ throw new IOException("Trying to load more than " + maxFilesPerRegionPerFamily +
+ " hfiles to one family of one region");
+ }
+
+ bulkLoadPhase(table, admin.getConnection(), pool, queue, regionGroups, copyFile,
+ item2RegionMap);
+
+ // NOTE: The next iteration's split / group could happen in parallel to
+ // atomic bulkloads assuming that there are splits and no merges, and
+ // that we can atomically pull out the groups we want to retry.
+ }
+
+ if (!queue.isEmpty()) {
+ throw new RuntimeException("Bulk load aborted with some files not yet loaded." +
+ "Please check log for more details.");
+ }
+ return item2RegionMap;
+ }
+
+ /**
+ * This takes the LQI's grouped by likely regions and attempts to bulk load them. Any failures are
+ * re-queued for another pass with the groupOrSplitPhase.
+ * <p>
+ * protected for testing.
+ */
+ @VisibleForTesting
+ protected void bulkLoadPhase(Table table, Connection conn, ExecutorService pool,
+ Deque<LoadQueueItem> queue, Multimap<ByteBuffer, LoadQueueItem> regionGroups,
+ boolean copyFile, Map<LoadQueueItem, ByteBuffer> item2RegionMap) throws IOException {
+ // atomically bulk load the groups.
+ Set<Future<List<LoadQueueItem>>> loadingFutures = new HashSet<>();
+ for (Entry<ByteBuffer, ? extends Collection<LoadQueueItem>> e : regionGroups.asMap()
+ .entrySet()) {
+ byte[] first = e.getKey().array();
+ Collection<LoadQueueItem> lqis = e.getValue();
+
+ ClientServiceCallable<byte[]> serviceCallable =
+ buildClientServiceCallable(conn, table.getName(), first, lqis, copyFile);
+
+ Callable<List<LoadQueueItem>> call = new Callable<List<LoadQueueItem>>() {
+ @Override
+ public List<LoadQueueItem> call() throws Exception {
+ List<LoadQueueItem> toRetry =
+ tryAtomicRegionLoad(serviceCallable, table.getName(), first, lqis);
+ return toRetry;
+ }
+ };
+ if (item2RegionMap != null) {
+ for (LoadQueueItem lqi : lqis) {
+ item2RegionMap.put(lqi, e.getKey());
+ }
+ }
+ loadingFutures.add(pool.submit(call));
+ }
+
+ // get all the results.
+ for (Future<List<LoadQueueItem>> future : loadingFutures) {
+ try {
+ List<LoadQueueItem> toRetry = future.get();
+
+ if (item2RegionMap != null) {
+ for (LoadQueueItem lqi : toRetry) {
+ item2RegionMap.remove(lqi);
+ }
+ }
+ // LQIs that are requeued to be regrouped.
+ queue.addAll(toRetry);
+
+ } catch (ExecutionException e1) {
+ Throwable t = e1.getCause();
+ if (t instanceof IOException) {
+ // At this point something unrecoverable has happened.
+ // TODO Implement bulk load recovery
+ throw new IOException("BulkLoad encountered an unrecoverable problem", t);
+ }
+ LOG.error("Unexpected execution exception during bulk load", e1);
+ throw new IllegalStateException(t);
+ } catch (InterruptedException e1) {
+ LOG.error("Unexpected interrupted exception during bulk load", e1);
+ throw (InterruptedIOException) new InterruptedIOException().initCause(e1);
+ }
+ }
+ }
+
+ @VisibleForTesting
+ protected ClientServiceCallable<byte[]> buildClientServiceCallable(Connection conn,
+ TableName tableName, byte[] first, Collection<LoadQueueItem> lqis, boolean copyFile) {
+ List<Pair<byte[], String>> famPaths =
+ lqis.stream().map(lqi -> Pair.newPair(lqi.getFamily(), lqi.getFilePath().toString()))
+ .collect(Collectors.toList());
+ return new ClientServiceCallable<byte[]>(conn, tableName, first,
+ rpcControllerFactory.newController(), HConstants.PRIORITY_UNSET) {
+ @Override
+ protected byte[] rpcCall() throws Exception {
+ SecureBulkLoadClient secureClient = null;
+ boolean success = false;
+ try {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Going to connect to server " + getLocation() + " for row " +
+ Bytes.toStringBinary(getRow()) + " with hfile group " +
+ LoadIncrementalHFiles.this.toString(famPaths));
+ }
+ byte[] regionName = getLocation().getRegionInfo().getRegionName();
+ try (Table table = conn.getTable(getTableName())) {
+ secureClient = new SecureBulkLoadClient(getConf(), table);
+ success = secureClient.secureBulkLoadHFiles(getStub(), famPaths, regionName,
+ assignSeqIds, fsDelegationToken.getUserToken(), bulkToken, copyFile);
+ }
+ return success ? regionName : null;
+ } finally {
+ // Best effort copying of files that might not have been imported
+ // from the staging directory back to original location
+ // in user directory
+ if (secureClient != null && !success) {
+ FileSystem targetFs = FileSystem.get(getConf());
+ FileSystem sourceFs = lqis.iterator().next().getFilePath().getFileSystem(getConf());
+ // Check to see if the source and target filesystems are the same
+ // If they are the same filesystem, we will try move the files back
+ // because previously we moved them to the staging directory.
+ if (FSHDFSUtils.isSameHdfs(getConf(), sourceFs, targetFs)) {
+ for (Pair<byte[], String> el : famPaths) {
+ Path hfileStagingPath = null;
+ Path hfileOrigPath = new Path(el.getSecond());
+ try {
+ hfileStagingPath = new Path(new Path(bulkToken, Bytes.toString(el.getFirst())),
+ hfileOrigPath.getName());
+ if (targetFs.rename(hfileStagingPath, hfileOrigPath)) {
+ LOG.debug("Moved back file " + hfileOrigPath + " from " + hfileStagingPath);
+ } else if (targetFs.exists(hfileStagingPath)) {
+ LOG.debug(
+ "Unable to move back file " + hfileOrigPath + " from " + hfileStagingPath);
+ }
+ } catch (Exception ex) {
+ LOG.debug(
+ "Unable to move back file " + hfileOrigPath + " from " + hfileStagingPath, ex);
+ }
+ }
+ }
+ }
+ }
+ }
+ };
+ }
+
+ private boolean checkHFilesCountPerRegionPerFamily(
+ final Multimap<ByteBuffer, LoadQueueItem> regionGroups) {
+ for (Map.Entry<ByteBuffer, Collection<LoadQueueItem>> e : regionGroups.asMap().entrySet()) {
+ Map<byte[], MutableInt> filesMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+ for (LoadQueueItem lqi : e.getValue()) {
+ MutableInt count = filesMap.computeIfAbsent(lqi.getFamily(), k -> new MutableInt());
+ count.increment();
+ if (count.intValue() > maxFilesPerRegionPerFamily) {
+ LOG.error("Trying to load more than " + maxFilesPerRegionPerFamily +
+ " hfiles to family " + Bytes.toStringBinary(lqi.getFamily()) +
+ " of region with start key " + Bytes.toStringBinary(e.getKey()));
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ /**
+ * @param table the table to load into
+ * @param pool the ExecutorService
+ * @param queue the queue for LoadQueueItem
+ * @param startEndKeys start and end keys
+ * @return A map that groups LQI by likely bulk load region targets and Set of missing hfiles.
+ */
+ private Pair<Multimap<ByteBuffer, LoadQueueItem>, Set<String>> groupOrSplitPhase(
+ final Table table, ExecutorService pool, Deque<LoadQueueItem> queue,
+ final Pair<byte[][], byte[][]> startEndKeys) throws IOException {
+ // <region start key, LQI> need synchronized only within this scope of this
+ // phase because of the puts that happen in futures.
+ Multimap<ByteBuffer, LoadQueueItem> rgs = HashMultimap.create();
+ final Multimap<ByteBuffer, LoadQueueItem> regionGroups = Multimaps.synchronizedMultimap(rgs);
+ Set<String> missingHFiles = new HashSet<>();
+ Pair<Multimap<ByteBuffer, LoadQueueItem>, Set<String>> pair =
+ new Pair<>(regionGroups, missingHFiles);
+
+ // drain LQIs and figure out bulk load groups
+ Set<Future<Pair<List<LoadQueueItem>, String>>> splittingFutures = new HashSet<>();
+ while (!queue.isEmpty()) {
+ final LoadQueueItem item = queue.remove();
+
+ final Callable<Pair<List<LoadQueueItem>, String>> call =
+ new Callable<Pair<List<LoadQueueItem>, String>>() {
+ @Override
+ public Pair<List<LoadQueueItem>, String> call() throws Exception {
+ Pair<List<LoadQueueItem>, String> splits =
+ groupOrSplit(regionGroups, item, table, startEndKeys);
+ return splits;
+ }
+ };
+ splittingFutures.add(pool.submit(call));
+ }
+ // get all the results. All grouping and splitting must finish before
+ // we can attempt the atomic loads.
+ for (Future<Pair<List<LoadQueueItem>, String>> lqis : splittingFutures) {
+ try {
+ Pair<List<LoadQueueItem>, String> splits = lqis.get();
+ if (splits != null) {
+ if (splits.getFirst() != null) {
+ queue.addAll(splits.getFirst());
+ } else {
+ missingHFiles.add(splits.getSecond());
+ }
+ }
+ } catch (ExecutionException e1) {
+ Throwable t = e1.getCause();
+ if (t instanceof IOException) {
+ LOG.error("IOException during splitting", e1);
+ throw (IOException) t; // would have been thrown if not parallelized,
+ }
+ LOG.error("Unexpected execution exception during splitting", e1);
+ throw new IllegalStateException(t);
+ } catch (InterruptedException e1) {
+ LOG.error("Unexpected interrupted exception during splitting", e1);
+ throw (InterruptedIOException) new InterruptedIOException().initCause(e1);
+ }
+ }
+ return pair;
+ }
+
+ private List<LoadQueueItem> splitStoreFile(final LoadQueueItem item, final Table table,
+ byte[] startKey, byte[] splitKey) throws IOException {
+ Path hfilePath = item.getFilePath();
+ byte[] family = item.getFamily();
+ Path tmpDir = hfilePath.getParent();
+ if (!tmpDir.getName().equals(TMP_DIR)) {
+ tmpDir = new Path(tmpDir, TMP_DIR);
+ }
+
+ LOG.info("HFile at " + hfilePath + " no longer fits inside a single " + "region. Splitting...");
+
+ String uniqueName = getUniqueName();
+ ColumnFamilyDescriptor familyDesc = table.getDescriptor().getColumnFamily(family);
+
+ Path botOut = new Path(tmpDir, uniqueName + ".bottom");
+ Path topOut = new Path(tmpDir, uniqueName + ".top");
+ splitStoreFile(getConf(), hfilePath, familyDesc, splitKey, botOut, topOut);
+
+ FileSystem fs = tmpDir.getFileSystem(getConf());
+ fs.setPermission(tmpDir, FsPermission.valueOf("-rwxrwxrwx"));
+ fs.setPermission(botOut, FsPermission.valueOf("-rwxrwxrwx"));
+ fs.setPermission(topOut, FsPermission.valueOf("-rwxrwxrwx"));
+
+ // Add these back at the *front* of the queue, so there's a lower
+ // chance that the region will just split again before we get there.
+ List<LoadQueueItem> lqis = new ArrayList<>(2);
+ lqis.add(new LoadQueueItem(family, botOut));
+ lqis.add(new LoadQueueItem(family, topOut));
+
+ // If the current item is already the result of previous splits,
+ // we don't need it anymore. Clean up to save space.
+ // It is not part of the original input files.
+ try {
+ if (tmpDir.getName().equals(TMP_DIR)) {
+ fs.delete(hfilePath, false);
+ }
+ } catch (IOException e) {
+ LOG.warn("Unable to delete temporary split file " + hfilePath);
+ }
+ LOG.info("Successfully split into new HFiles " + botOut + " and " + topOut);
+ return lqis;
+ }
+
+ /**
+ * Attempt to assign the given load queue item into its target region group. If the hfile boundary
+ * no longer fits into a region, physically splits the hfile such that the new bottom half will
+ * fit and returns the list of LQI's corresponding to the resultant hfiles.
+ * <p>
+ * protected for testing
+ * @throws IOException if an IO failure is encountered
+ */
+ @VisibleForTesting
+ protected Pair<List<LoadQueueItem>, String> groupOrSplit(
+ Multimap<ByteBuffer, LoadQueueItem> regionGroups, final LoadQueueItem item, final Table table,
+ final Pair<byte[][], byte[][]> startEndKeys) throws IOException {
+ Path hfilePath = item.getFilePath();
+ byte[] first, last;
+ try (HFile.Reader hfr = HFile.createReader(hfilePath.getFileSystem(getConf()), hfilePath,
+ new CacheConfig(getConf()), true, getConf())) {
+ hfr.loadFileInfo();
+ first = hfr.getFirstRowKey();
+ last = hfr.getLastRowKey();
+ } catch (FileNotFoundException fnfe) {
+ LOG.debug("encountered", fnfe);
+ return new Pair<>(null, hfilePath.getName());
+ }
+
+ LOG.info("Trying to load hfile=" + hfilePath + " first=" + Bytes.toStringBinary(first) +
+ " last=" + Bytes.toStringBinary(last));
+ if (first == null || last == null) {
+ assert first == null && last == null;
+ // TODO what if this is due to a bad HFile?
+ LOG.info("hfile " + hfilePath + " has no entries, skipping");
+ return null;
+ }
+ if (Bytes.compareTo(first, last) > 0) {
+ throw new IllegalArgumentException(
+ "Invalid range: " + Bytes.toStringBinary(first) + " > " + Bytes.toStringBinary(last));
+ }
+ int idx = Arrays.binarySearch(startEndKeys.getFirst(), first, Bytes.BYTES_COMPARATOR);
+ if (idx < 0) {
+ // not on boundary, returns -(insertion index). Calculate region it
+ // would be in.
+ idx = -(idx + 1) - 1;
+ }
+ int indexForCallable = idx;
+
+ /**
+ * we can consider there is a region hole in following conditions. 1) if idx < 0,then first
+ * region info is lost. 2) if the endkey of a region is not equal to the startkey of the next
+ * region. 3) if the endkey of the last region is not empty.
+ */
+ if (indexForCallable < 0) {
+ throw new IOException("The first region info for table " + table.getName() +
+ " can't be found in hbase:meta.Please use hbck tool to fix it first.");
+ } else if ((indexForCallable == startEndKeys.getFirst().length - 1) &&
+ !Bytes.equals(startEndKeys.getSecond()[indexForCallable], HConstants.EMPTY_BYTE_ARRAY)) {
+ throw new IOException("The last region info for table " + table.getName() +
+ " can't be found in hbase:meta.Please use hbck tool to fix it first.");
+ } else if (indexForCallable + 1 < startEndKeys.getFirst().length &&
+ !(Bytes.compareTo(startEndKeys.getSecond()[indexForCallable],
+ startEndKeys.getFirst()[indexForCallable + 1]) == 0)) {
+ throw new IOException("The endkey of one region for table " + table.getName() +
+ " is not equal to the startkey of the next region in hbase:meta." +
+ "Please use hbck tool to fix it first.");
+ }
+
+ boolean lastKeyInRange = Bytes.compareTo(last, startEndKeys.getSecond()[idx]) < 0 ||
+ Bytes.equals(startEndKeys.getSecond()[idx], HConstants.EMPTY_BYTE_ARRAY);
+ if (!lastKeyInRange) {
+ List<LoadQueueItem> lqis = splitStoreFile(item, table,
+ startEndKeys.getFirst()[indexForCallable], startEndKeys.getSecond()[indexForCallable]);
+ return new Pair<>(lqis, null);
+ }
+
+ // group regions.
+ regionGroups.put(ByteBuffer.wrap(startEndKeys.getFirst()[idx]), item);
+ return null;
+ }
+
+ /**
+ * Attempts to do an atomic load of many hfiles into a region. If it fails, it returns a list of
+ * hfiles that need to be retried. If it is successful it will return an empty list.
+ * <p>
+ * NOTE: To maintain row atomicity guarantees, region server callable should succeed atomically
+ * and fails atomically.
+ * <p>
+ * Protected for testing.
+ * @return empty list if success, list of items to retry on recoverable failure
+ */
+ @VisibleForTesting
+ protected List<LoadQueueItem> tryAtomicRegionLoad(ClientServiceCallable<byte[]> serviceCallable,
+ final TableName tableName, final byte[] first, final Collection<LoadQueueItem> lqis)
+ throws IOException {
+ try {
+ List<LoadQueueItem> toRetry = new ArrayList<>();
+ Configuration conf = getConf();
+ byte[] region = RpcRetryingCallerFactory.instantiate(conf, null).<byte[]> newCaller()
+ .callWithRetries(serviceCallable, Integer.MAX_VALUE);
+ if (region == null) {
+ LOG.warn("Attempt to bulk load region containing " + Bytes.toStringBinary(first) +
+ " into table " + tableName + " with files " + lqis +
+ " failed. This is recoverable and they will be retried.");
+ toRetry.addAll(lqis); // return lqi's to retry
+ }
+ // success
+ return toRetry;
+ } catch (IOException e) {
+ LOG.error("Encountered unrecoverable error from region server, additional details: " +
+ serviceCallable.getExceptionMessageAdditionalDetail(),
+ e);
+ throw e;
+ }
+ }
+
+ /**
+ * If the table is created for the first time, then "completebulkload" reads the files twice. More
+ * modifications necessary if we want to avoid doing it.
+ */
+ private void createTable(TableName tableName, String dirPath, Admin admin) throws IOException {
+ final Path hfofDir = new Path(dirPath);
+ final FileSystem fs = hfofDir.getFileSystem(getConf());
+
+ // Add column families
+ // Build a set of keys
+ List<ColumnFamilyDescriptorBuilder> familyBuilders = new ArrayList<>();
+ SortedMap<byte[], Integer> map = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+ visitBulkHFiles(fs, hfofDir, new BulkHFileVisitor<ColumnFamilyDescriptorBuilder>() {
+ @Override
+ public ColumnFamilyDescriptorBuilder bulkFamily(byte[] familyName) {
+ ColumnFamilyDescriptorBuilder builder =
+ ColumnFamilyDescriptorBuilder.newBuilder(familyName);
+ familyBuilders.add(builder);
+ return builder;
+ }
+
+ @Override
+ public void bulkHFile(ColumnFamilyDescriptorBuilder builder, FileStatus hfileStatus)
+ throws IOException {
+ Path hfile = hfileStatus.getPath();
+ try (HFile.Reader reader =
+ HFile.createReader(fs, hfile, new CacheConfig(getConf()), true, getConf())) {
+ if (builder.getCompressionType() != reader.getFileContext().getCompression()) {
+ builder.setCompressionType(reader.getFileContext().getCompression());
+ LOG.info("Setting compression " + reader.getFileContext().getCompression().name() +
+ " for family " + builder.getNameAsString());
+ }
+ reader.loadFileInfo();
+ byte[] first = reader.getFirstRowKey();
+ byte[] last = reader.getLastRowKey();
+
+ LOG.info("Trying to figure out region boundaries hfile=" + hfile + " first=" +
+ Bytes.toStringBinary(first) + " last=" + Bytes.toStringBinary(last));
+
+ // To eventually infer start key-end key boundaries
+ Integer value = map.containsKey(first) ? map.get(first) : 0;
+ map.put(first, value + 1);
+
+ value = map.containsKey(last) ? map.get(last) : 0;
+ map.put(last, value - 1);
+ }
+ }
+ });
+
+ byte[][] keys = inferBoundaries(map);
+ TableDescriptorBuilder tdBuilder = TableDescriptorBuilder.newBuilder(tableName);
+ familyBuilders.stream().map(ColumnFamilyDescriptorBuilder::build)
+ .forEachOrdered(tdBuilder::addColumnFamily);
+ admin.createTable(tdBuilder.build(), keys);
+
+ LOG.info("Table " + tableName + " is available!!");
+ }
+
+ private void cleanup(Admin admin, Deque<LoadQueueItem> queue, ExecutorService pool,
+ SecureBulkLoadClient secureClient) throws IOException {
+ fsDelegationToken.releaseDelegationToken();
+ if (bulkToken != null && secureClient != null) {
+ secureClient.cleanupBulkLoad(admin.getConnection(), bulkToken);
+ }
+ if (pool != null) {
+ pool.shutdown();
+ }
+ if (!queue.isEmpty()) {
+ StringBuilder err = new StringBuilder();
+ err.append("-------------------------------------------------\n");
+ err.append("Bulk load aborted with some files not yet loaded:\n");
+ err.append("-------------------------------------------------\n");
+ for (LoadQueueItem q : queue) {
+ err.append(" ").append(q.getFilePath()).append('\n');
+ }
+ LOG.error(err);
+ }
+ }
+
+ // unique file name for the table
+ private String getUniqueName() {
+ return UUID.randomUUID().toString().replaceAll("-", "");
+ }
+
+ /**
+ * Checks whether there is any invalid family name in HFiles to be bulk loaded.
+ */
+ private void validateFamiliesInHFiles(Table table, Deque<LoadQueueItem> queue, boolean silence)
+ throws IOException {
+ Set<String> familyNames = Arrays.asList(table.getDescriptor().getColumnFamilies()).stream()
+ .map(f -> f.getNameAsString()).collect(Collectors.toSet());
+ List<String> unmatchedFamilies = queue.stream().map(item -> Bytes.toString(item.getFamily()))
+ .filter(fn -> !familyNames.contains(fn)).distinct().collect(Collectors.toList());
+ if (unmatchedFamilies.size() > 0) {
+ String msg =
+ "Unmatched family names found: unmatched family names in HFiles to be bulkloaded: " +
+ unmatchedFamilies + "; valid family names of table " + table.getName() + " are: " +
+ familyNames;
+ LOG.error(msg);
+ if (!silence) {
+ throw new IOException(msg);
+ }
+ }
+ }
+
+ /**
+ * Populate the Queue with given HFiles
+ */
+ private void populateLoadQueue(Deque<LoadQueueItem> ret, Map<byte[], List<Path>> map) {
+ map.forEach((k, v) -> v.stream().map(p -> new LoadQueueItem(k, p)).forEachOrdered(ret::add));
+ }
+
+ /**
+ * Walk the given directory for all HFiles, and return a Queue containing all such files.
+ */
+ private void discoverLoadQueue(final Deque<LoadQueueItem> ret, final Path hfofDir,
+ final boolean validateHFile) throws IOException {
+ visitBulkHFiles(hfofDir.getFileSystem(getConf()), hfofDir, new BulkHFileVisitor<byte[]>() {
+ @Override
+ public byte[] bulkFamily(final byte[] familyName) {
+ return familyName;
+ }
+
+ @Override
+ public void bulkHFile(final byte[] family, final FileStatus hfile) throws IOException {
+ long length = hfile.getLen();
+ if (length > getConf().getLong(HConstants.HREGION_MAX_FILESIZE,
+ HConstants.DEFAULT_MAX_FILE_SIZE)) {
+ LOG.warn("Trying to bulk load hfile " + hfile.getPath() + " with size: " + length +
+ " bytes can be problematic as it may lead to oversplitting.");
+ }
+ ret.add(new LoadQueueItem(family, hfile.getPath()));
+ }
+ }, validateHFile);
+ }
+
+ private interface BulkHFileVisitor<TFamily> {
+
+ TFamily bulkFamily(byte[] familyName) throws IOException;
+
+ void bulkHFile(TFamily family, FileStatus hfileStatus) throws IOException;
+ }
+
+ /**
+ * Iterate over the bulkDir hfiles. Skip reference, HFileLink, files starting with "_" and
+ * non-valid hfiles.
+ */
+ private static <TFamily> void visitBulkHFiles(final FileSystem fs, final Path bulkDir,
+ final BulkHFileVisitor<TFamily> visitor) throws IOException {
+ visitBulkHFiles(fs, bulkDir, visitor, true);
+ }
+
+ /**
+ * Iterate over the bulkDir hfiles. Skip reference, HFileLink, files starting with "_". Check and
+ * skip non-valid hfiles by default, or skip this validation by setting
+ * 'hbase.loadincremental.validate.hfile' to false.
+ */
+ private static <TFamily> void visitBulkHFiles(FileSystem fs, Path bulkDir,
+ BulkHFileVisitor<TFamily> visitor, boolean validateHFile) throws IOException {
+ FileStatus[] familyDirStatuses = fs.listStatus(bulkDir);
+ for (FileStatus familyStat : familyDirStatuses) {
+ if (!familyStat.isDirectory()) {
+ LOG.warn("Skipping non-directory " + familyStat.getPath());
+ continue;
+ }
+ Path familyDir = familyStat.getPath();
+ byte[] familyName = familyDir.getName().getBytes();
+ // Skip invalid family
+ try {
+ ColumnFamilyDescriptorBuilder.isLegalColumnFamilyName(familyName);
+ } catch (IllegalArgumentException e) {
+ LOG.warn("Skipping invalid " + familyStat.getPath());
+ continue;
+ }
+ TFamily family = visitor.bulkFamily(familyName);
+
+ FileStatus[] hfileStatuses = fs.listStatus(familyDir);
+ for (FileStatus hfileStatus : hfileStatuses) {
+ if (!fs.isFile(hfileStatus.getPath())) {
+ LOG.warn("Skipping non-file " + hfileStatus);
+ continue;
+ }
+
+ Path hfile = hfileStatus.getPath();
+ // Skip "_", reference, HFileLink
+ String fileName = hfile.getName();
+ if (fileName.startsWith("_")) {
+ continue;
+ }
+ if (StoreFileInfo.isReference(fileName)) {
+ LOG.warn("Skipping reference " + fileName);
+ continue;
+ }
+ if (HFileLink.isHFileLink(fileName)) {
+ LOG.warn("Skipping HFileLink " + fileName);
+ continue;
+ }
+
+ // Validate HFile Format if needed
+ if (validateHFile) {
+ try {
+ if (!HFile.isHFileFormat(fs, hfile)) {
+ LOG.warn("the file " + hfile + " doesn't seems to be an hfile. skipping");
+ continue;
+ }
+ } catch (FileNotFoundException e) {
+ LOG.warn("the file " + hfile + " was removed");
+ continue;
+ }
+ }
+
+ visitor.bulkHFile(family, hfileStatus);
+ }
+ }
+ }
+
+ // Initialize a thread pool
+ private ExecutorService createExecutorService() {
+ ThreadPoolExecutor pool = new ThreadPoolExecutor(nrThreads, nrThreads, 60, TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ new ThreadFactoryBuilder().setNameFormat("LoadIncrementalHFiles-%1$d").build());
+ pool.allowCoreThreadTimeOut(true);
+ return pool;
+ }
+
+ private final String toString(List<Pair<byte[], String>> list) {
+ StringBuilder sb = new StringBuilder();
+ sb.append('[');
+ list.forEach(p -> {
+ sb.append('{').append(Bytes.toStringBinary(p.getFirst())).append(',').append(p.getSecond())
+ .append('}');
+ });
+ sb.append(']');
+ return sb.toString();
+ }
+
+ private boolean isSecureBulkLoadEndpointAvailable() {
+ String classes = getConf().get(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY, "");
+ return classes.contains("org.apache.hadoop.hbase.security.access.SecureBulkLoadEndpoint");
+ }
+
+ /**
+ * Split a storefile into a top and bottom half, maintaining the metadata, recreating bloom
+ * filters, etc.
+ */
+ @VisibleForTesting
+ static void splitStoreFile(Configuration conf, Path inFile, ColumnFamilyDescriptor familyDesc,
+ byte[] splitKey, Path bottomOut, Path topOut) throws IOException {
+ // Open reader with no block cache, and not in-memory
+ Reference topReference = Reference.createTopReference(splitKey);
+ Reference bottomReference = Reference.createBottomReference(splitKey);
+
+ copyHFileHalf(conf, inFile, topOut, topReference, familyDesc);
+ copyHFileHalf(conf, inFile, bottomOut, bottomReference, familyDesc);
+ }
+
+ /**
+ * Copy half of an HFile into a new HFile.
+ */
+ private static void copyHFileHalf(Configuration conf, Path inFile, Path outFile,
+ Reference reference, ColumnFamilyDescriptor familyDescriptor) throws IOException {
+ FileSystem fs = inFile.getFileSystem(conf);
+ CacheConfig cacheConf = new CacheConfig(conf);
+ HalfStoreFileReader halfReader = null;
+ StoreFileWriter halfWriter = null;
+ try {
+ halfReader = new HalfStoreFileReader(fs, inFile, cacheConf, reference, true,
+ new AtomicInteger(0), true, conf);
+ Map<byte[], byte[]> fileInfo = halfReader.loadFileInfo();
+
+ int blocksize = familyDescriptor.getBlocksize();
+ Algorithm compression = familyDescriptor.getCompressionType();
+ BloomType bloomFilterType = familyDescriptor.getBloomFilterType();
+ HFileContext hFileContext = new HFileContextBuilder().withCompression(compression)
+ .withChecksumType(HStore.getChecksumType(conf))
+ .withBytesPerCheckSum(HStore.getBytesPerChecksum(conf)).withBlockSize(blocksize)
+ .withDataBlockEncoding(familyDescriptor.getDataBlockEncoding()).withIncludesTags(true)
+ .build();
+ halfWriter = new StoreFileWriter.Builder(conf, cacheConf, fs).withFilePath(outFile)
+ .withBloomType(bloomFilterType).withFileContext(hFileContext).build();
+ HFileScanner scanner = halfReader.getScanner(false, false, false);
+ scanner.seekTo();
+ do {
+ halfWriter.append(scanner.getCell());
+ } while (scanner.next());
+
+ for (Map.Entry<byte[], byte[]> entry : fileInfo.entrySet()) {
+ if (shouldCopyHFileMetaKey(entry.getKey())) {
+ halfWriter.appendFileInfo(entry.getKey(), entry.getValue());
+ }
+ }
+ } finally {
+ if (halfReader != null) {
+ try {
+ halfReader.close(cacheConf.shouldEvictOnClose());
+ } catch (IOException e) {
+ LOG.warn("failed to close hfile reader for " + inFile, e);
+ }
+ }
+ if (halfWriter != null) {
+ halfWriter.close();
+ }
+
+ }
+ }
+
+ private static boolean shouldCopyHFileMetaKey(byte[] key) {
+ // skip encoding to keep hfile meta consistent with data block info, see HBASE-15085
+ if (Bytes.equals(key, HFileDataBlockEncoder.DATA_BLOCK_ENCODING)) {
+ return false;
+ }
+
+ return !HFile.isReservedFileInfoKey(key);
+ }
+
+ private boolean isCreateTable() {
+ return "yes".equalsIgnoreCase(getConf().get(CREATE_TABLE_CONF_KEY, "yes"));
+ }
+
+ private boolean isSilence() {
+ return "yes".equalsIgnoreCase(getConf().get(IGNORE_UNMATCHED_CF_CONF_KEY, ""));
+ }
+
+ private boolean isAlwaysCopyFiles() {
+ return getConf().getBoolean(ALWAYS_COPY_FILES, false);
+ }
+
+ /**
+ * Perform bulk load on the given table.
+ * @param hfofDir the directory that was provided as the output path of a job using
+ * HFileOutputFormat
+ * @param tableName the table to load into
+ */
+ public Map<LoadQueueItem, ByteBuffer> run(String hfofDir, TableName tableName)
+ throws IOException {
+ try (Connection connection = ConnectionFactory.createConnection(getConf());
+ Admin admin = connection.getAdmin()) {
+ if (!admin.tableExists(tableName)) {
+ if (isCreateTable()) {
+ createTable(tableName, hfofDir, admin);
+ } else {
+ String errorMsg = format("Table '%s' does not exist.", tableName);
+ LOG.error(errorMsg);
+ throw new TableNotFoundException(errorMsg);
+ }
+ }
+ try (Table table = connection.getTable(tableName);
+ RegionLocator locator = connection.getRegionLocator(tableName)) {
+ return doBulkLoad(new Path(hfofDir), admin, table, locator, isSilence(), isAlwaysCopyFiles());
+ }
+ }
+ }
+
+ /**
+ * Perform bulk load on the given table.
+ * @param family2Files map of family to List of hfiles
+ * @param tableName the table to load into
+ */
+ public Map<LoadQueueItem, ByteBuffer> run(Map<byte[], List<Path>> family2Files,
+ TableName tableName) throws IOException {
+ try (Connection connection = ConnectionFactory.createConnection(getConf());
+ Admin admin = connection.getAdmin()) {
+ if (!admin.tableExists(tableName)) {
+ String errorMsg = format("Table '%s' does not exist.", tableName);
+ LOG.error(errorMsg);
+ throw new TableNotFoundException(errorMsg);
+ }
+ try (Table table = connection.getTable(tableName);
+ RegionLocator locator = connection.getRegionLocator(tableName)) {
+ return doBulkLoad(family2Files, admin, table, locator, isSilence(), isAlwaysCopyFiles());
+ }
+ }
+ }
+
+ @Override
+ public int run(String[] args) throws Exception {
+ if (args.length < 2) {
+ usage();
+ return -1;
+ }
+ String dirPath = args[0];
+ TableName tableName = TableName.valueOf(args[1]);
+ return !run(dirPath, tableName).isEmpty() ? 0 : -1;
+ }
+
+ public static void main(String[] args) throws Exception {
+ Configuration conf = HBaseConfiguration.create();
+ int ret = ToolRunner.run(conf, new LoadIncrementalHFiles(conf), args);
+ System.exit(ret);
+ }
+
+ /**
+ * Called from replication sink, where it manages bulkToken(staging directory) by itself. This is
+ * used only when SecureBulkLoadEndpoint is configured in hbase.coprocessor.region.classes
+ * property. This directory is used as a temporary directory where all files are initially
+ * copied/moved from user given directory, set all the required file permissions and then from
+ * their it is finally loaded into a table. This should be set only when, one would like to manage
+ * the staging directory by itself. Otherwise this tool will handle this by itself.
+ * @param stagingDir staging directory path
+ */
+ public void setBulkToken(String stagingDir) {
+ this.bulkToken = stagingDir;
+ }
+
+ /**
+ * Infers region boundaries for a new table.
+ * <p>
+ * Parameter: <br>
+ * bdryMap is a map between keys to an integer belonging to {+1, -1}
+ * <ul>
+ * <li>If a key is a start key of a file, then it maps to +1</li>
+ * <li>If a key is an end key of a file, then it maps to -1</li>
+ * </ul>
+ * <p>
+ * Algo:<br>
+ * <ol>
+ * <li>Poll on the keys in order:
+ * <ol type="a">
+ * <li>Keep adding the mapped values to these keys (runningSum)</li>
+ * <li>Each time runningSum reaches 0, add the start Key from when the runningSum had started to a
+ * boundary list.</li>
+ * </ol>
+ * </li>
+ * <li>Return the boundary list.</li>
+ * </ol>
+ */
+ public static byte[][] inferBoundaries(SortedMap<byte[], Integer> bdryMap) {
+ List<byte[]> keysArray = new ArrayList<>();
+ int runningValue = 0;
+ byte[] currStartKey = null;
+ boolean firstBoundary = true;
+
+ for (Map.Entry<byte[], Integer> item : bdryMap.entrySet()) {
+ if (runningValue == 0) {
+ currStartKey = item.getKey();
+ }
+ runningValue += item.getValue();
+ if (runningValue == 0) {
+ if (!firstBoundary) {
+ keysArray.add(currStartKey);
+ }
+ firstBoundary = false;
+ }
+ }
+
+ return keysArray.toArray(new byte[0][]);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hbase/blob/9e53f292/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
index bc663e1..886d0da 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/util/HBaseFsck.java
@@ -3484,7 +3484,7 @@ public class HBaseFsck extends Configured implements Closeable {
errors.print("This sidelined region dir should be bulk loaded: "
+ path.toString());
errors.print("Bulk load command looks like: "
- + "hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles "
+ + "hbase org.apache.hadoop.hbase.tool.LoadIncrementalHFiles "
+ path.toUri().getPath() + " "+ tableName);
}
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/9e53f292/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
index c4924bb..3b7f1f8 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
@@ -63,7 +63,6 @@ import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileContext;
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
-import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.NoLimitScannerContext;
@@ -72,9 +71,9 @@ import org.apache.hadoop.hbase.regionserver.ScanType;
import org.apache.hadoop.hbase.regionserver.ScannerContext;
import org.apache.hadoop.hbase.regionserver.Store;
import org.apache.hadoop.hbase.regionserver.StoreFile;
-import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.testclassification.CoprocessorTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
@@ -86,6 +85,8 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.rules.TestName;
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+
@Category({ CoprocessorTests.class, MediumTests.class })
public class TestRegionObserverInterface {
private static final Log LOG = LogFactory.getLog(TestRegionObserverInterface.class);
http://git-wip-us.apache.org/repos/asf/hbase/blob/9e53f292/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
deleted file mode 100644
index b5b7a0c..0000000
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/mapreduce/TestLoadIncrementalHFiles.java
+++ /dev/null
@@ -1,763 +0,0 @@
-/**
- *
- * 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.hadoop.hbase.mapreduce;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.TreeMap;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-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.NamespaceDescriptor;
-import org.apache.hadoop.hbase.TableName;
-import org.apache.hadoop.hbase.TableNotFoundException;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.codec.KeyValueCodecWithTags;
-import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
-import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
-import org.apache.hadoop.hbase.io.hfile.CacheConfig;
-import org.apache.hadoop.hbase.io.hfile.HFile;
-import org.apache.hadoop.hbase.io.hfile.HFileScanner;
-import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles.LoadQueueItem;
-import org.apache.hadoop.hbase.regionserver.BloomType;
-import org.apache.hadoop.hbase.testclassification.LargeTests;
-import org.apache.hadoop.hbase.testclassification.MapReduceTests;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.FSUtils;
-import org.apache.hadoop.hbase.util.HFileTestUtil;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.rules.TestName;
-
-/**
- * Test cases for the "load" half of the HFileOutputFormat bulk load
- * functionality. These tests run faster than the full MR cluster
- * tests in TestHFileOutputFormat
- */
-@Category({MapReduceTests.class, LargeTests.class})
-public class TestLoadIncrementalHFiles {
- @Rule
- public TestName tn = new TestName();
-
- private static final byte[] QUALIFIER = Bytes.toBytes("myqual");
- private static final byte[] FAMILY = Bytes.toBytes("myfam");
- private static final String NAMESPACE = "bulkNS";
-
- static final String EXPECTED_MSG_FOR_NON_EXISTING_FAMILY = "Unmatched family names found";
- static final int MAX_FILES_PER_REGION_PER_FAMILY = 4;
-
- private static final byte[][] SPLIT_KEYS = new byte[][] {
- Bytes.toBytes("ddd"),
- Bytes.toBytes("ppp")
- };
-
- static HBaseTestingUtility util = new HBaseTestingUtility();
-
- @BeforeClass
- public static void setUpBeforeClass() throws Exception {
- util.getConfiguration().set(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,"");
- util.getConfiguration().setInt(
- LoadIncrementalHFiles.MAX_FILES_PER_REGION_PER_FAMILY,
- MAX_FILES_PER_REGION_PER_FAMILY);
- // change default behavior so that tag values are returned with normal rpcs
- util.getConfiguration().set(HConstants.RPC_CODEC_CONF_KEY,
- KeyValueCodecWithTags.class.getCanonicalName());
- util.startMiniCluster();
-
- setupNamespace();
- }
-
- protected static void setupNamespace() throws Exception {
- util.getAdmin().createNamespace(NamespaceDescriptor.create(NAMESPACE).build());
- }
-
- @AfterClass
- public static void tearDownAfterClass() throws Exception {
- util.shutdownMiniCluster();
- }
-
- @Test(timeout = 120000)
- public void testSimpleLoadWithMap() throws Exception {
- runTest("testSimpleLoadWithMap", BloomType.NONE,
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("cccc") },
- new byte[][]{ Bytes.toBytes("ddd"), Bytes.toBytes("ooo") },
- }, true);
- }
-
- /**
- * Test case that creates some regions and loads
- * HFiles that fit snugly inside those regions
- */
- @Test(timeout = 120000)
- public void testSimpleLoad() throws Exception {
- runTest("testSimpleLoad", BloomType.NONE,
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("cccc") },
- new byte[][]{ Bytes.toBytes("ddd"), Bytes.toBytes("ooo") },
- });
- }
-
- @Test(timeout = 120000)
- public void testSimpleLoadWithFileCopy() throws Exception {
- String testName = tn.getMethodName();
- final byte[] TABLE_NAME = Bytes.toBytes("mytable_" + testName);
- runTest(testName, buildHTD(TableName.valueOf(TABLE_NAME), BloomType.NONE), BloomType.NONE,
- false, null, new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("cccc") },
- new byte[][]{ Bytes.toBytes("ddd"), Bytes.toBytes("ooo") },
- }, false, true);
- }
-
- /**
- * Test case that creates some regions and loads
- * HFiles that cross the boundaries of those regions
- */
- @Test(timeout = 120000)
- public void testRegionCrossingLoad() throws Exception {
- runTest("testRegionCrossingLoad", BloomType.NONE,
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("eee") },
- new byte[][]{ Bytes.toBytes("fff"), Bytes.toBytes("zzz") },
- });
- }
-
- /**
- * Test loading into a column family that has a ROW bloom filter.
- */
- @Test(timeout = 60000)
- public void testRegionCrossingRowBloom() throws Exception {
- runTest("testRegionCrossingLoadRowBloom", BloomType.ROW,
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("eee") },
- new byte[][]{ Bytes.toBytes("fff"), Bytes.toBytes("zzz") },
- });
- }
-
- /**
- * Test loading into a column family that has a ROWCOL bloom filter.
- */
- @Test(timeout = 120000)
- public void testRegionCrossingRowColBloom() throws Exception {
- runTest("testRegionCrossingLoadRowColBloom", BloomType.ROWCOL,
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("eee") },
- new byte[][]{ Bytes.toBytes("fff"), Bytes.toBytes("zzz") },
- });
- }
-
- /**
- * Test case that creates some regions and loads HFiles that have
- * different region boundaries than the table pre-split.
- */
- @Test(timeout = 120000)
- public void testSimpleHFileSplit() throws Exception {
- runTest("testHFileSplit", BloomType.NONE,
- new byte[][] {
- Bytes.toBytes("aaa"), Bytes.toBytes("fff"), Bytes.toBytes("jjj"),
- Bytes.toBytes("ppp"), Bytes.toBytes("uuu"), Bytes.toBytes("zzz"),
- },
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("lll") },
- new byte[][]{ Bytes.toBytes("mmm"), Bytes.toBytes("zzz") },
- }
- );
- }
-
- /**
- * Test case that creates some regions and loads HFiles that cross the boundaries
- * and have different region boundaries than the table pre-split.
- */
- @Test(timeout = 60000)
- public void testRegionCrossingHFileSplit() throws Exception {
- testRegionCrossingHFileSplit(BloomType.NONE);
- }
-
- /**
- * Test case that creates some regions and loads HFiles that cross the boundaries
- * have a ROW bloom filter and a different region boundaries than the table pre-split.
- */
- @Test(timeout = 120000)
- public void testRegionCrossingHFileSplitRowBloom() throws Exception {
- testRegionCrossingHFileSplit(BloomType.ROW);
- }
-
- /**
- * Test case that creates some regions and loads HFiles that cross the boundaries
- * have a ROWCOL bloom filter and a different region boundaries than the table pre-split.
- */
- @Test(timeout = 120000)
- public void testRegionCrossingHFileSplitRowColBloom() throws Exception {
- testRegionCrossingHFileSplit(BloomType.ROWCOL);
- }
-
- @Test
- public void testSplitALot() throws Exception {
- runTest("testSplitALot", BloomType.NONE,
- new byte[][] {
- Bytes.toBytes("aaaa"), Bytes.toBytes("bbb"),
- Bytes.toBytes("ccc"), Bytes.toBytes("ddd"),
- Bytes.toBytes("eee"), Bytes.toBytes("fff"),
- Bytes.toBytes("ggg"), Bytes.toBytes("hhh"),
- Bytes.toBytes("iii"), Bytes.toBytes("lll"),
- Bytes.toBytes("mmm"), Bytes.toBytes("nnn"),
- Bytes.toBytes("ooo"), Bytes.toBytes("ppp"),
- Bytes.toBytes("qqq"), Bytes.toBytes("rrr"),
- Bytes.toBytes("sss"), Bytes.toBytes("ttt"),
- Bytes.toBytes("uuu"), Bytes.toBytes("vvv"),
- Bytes.toBytes("zzz"),
- },
- new byte[][][] {
- new byte[][] { Bytes.toBytes("aaaa"), Bytes.toBytes("zzz") },
- }
- );
- }
-
- private void testRegionCrossingHFileSplit(BloomType bloomType) throws Exception {
- runTest("testHFileSplit" + bloomType + "Bloom", bloomType,
- new byte[][] {
- Bytes.toBytes("aaa"), Bytes.toBytes("fff"), Bytes.toBytes("jjj"),
- Bytes.toBytes("ppp"), Bytes.toBytes("uuu"), Bytes.toBytes("zzz"),
- },
- new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaaa"), Bytes.toBytes("eee") },
- new byte[][]{ Bytes.toBytes("fff"), Bytes.toBytes("zzz") },
- }
- );
- }
-
- private HTableDescriptor buildHTD(TableName tableName, BloomType bloomType) {
- HTableDescriptor htd = new HTableDescriptor(tableName);
- HColumnDescriptor familyDesc = new HColumnDescriptor(FAMILY);
- familyDesc.setBloomFilterType(bloomType);
- htd.addFamily(familyDesc);
- return htd;
- }
-
- private void runTest(String testName, BloomType bloomType,
- byte[][][] hfileRanges) throws Exception {
- runTest(testName, bloomType, null, hfileRanges);
- }
-
- private void runTest(String testName, BloomType bloomType,
- byte[][][] hfileRanges, boolean useMap) throws Exception {
- runTest(testName, bloomType, null, hfileRanges, useMap);
- }
-
- private void runTest(String testName, BloomType bloomType,
- byte[][] tableSplitKeys, byte[][][] hfileRanges) throws Exception {
- runTest(testName, bloomType, tableSplitKeys, hfileRanges, false);
- }
-
- private void runTest(String testName, BloomType bloomType,
- byte[][] tableSplitKeys, byte[][][] hfileRanges, boolean useMap) throws Exception {
- final byte[] TABLE_NAME = Bytes.toBytes("mytable_"+testName);
- final boolean preCreateTable = tableSplitKeys != null;
-
- // Run the test bulkloading the table to the default namespace
- final TableName TABLE_WITHOUT_NS = TableName.valueOf(TABLE_NAME);
- runTest(testName, TABLE_WITHOUT_NS, bloomType, preCreateTable, tableSplitKeys, hfileRanges,
- useMap);
-
- // Run the test bulkloading the table to the specified namespace
- final TableName TABLE_WITH_NS = TableName.valueOf(Bytes.toBytes(NAMESPACE), TABLE_NAME);
- runTest(testName, TABLE_WITH_NS, bloomType, preCreateTable, tableSplitKeys, hfileRanges,
- useMap);
- }
-
- private void runTest(String testName, TableName tableName, BloomType bloomType,
- boolean preCreateTable, byte[][] tableSplitKeys, byte[][][] hfileRanges, boolean useMap)
- throws Exception {
- HTableDescriptor htd = buildHTD(tableName, bloomType);
- runTest(testName, htd, bloomType, preCreateTable, tableSplitKeys, hfileRanges, useMap, false);
- }
-
- public static int loadHFiles(String testName, HTableDescriptor htd, HBaseTestingUtility util,
- byte[] fam, byte[] qual, boolean preCreateTable, byte[][] tableSplitKeys,
- byte[][][] hfileRanges, boolean useMap, boolean deleteFile,
- boolean copyFiles, int initRowCount, int factor) throws Exception {
- Path dir = util.getDataTestDirOnTestFS(testName);
- FileSystem fs = util.getTestFileSystem();
- dir = dir.makeQualified(fs);
- Path familyDir = new Path(dir, Bytes.toString(fam));
-
- int hfileIdx = 0;
- Map<byte[], List<Path>> map = null;
- List<Path> list = null;
- if (useMap || copyFiles) {
- list = new ArrayList<>();
- }
- if (useMap) {
- map = new TreeMap<>(Bytes.BYTES_COMPARATOR);
- map.put(fam, list);
- }
- Path last = null;
- for (byte[][] range : hfileRanges) {
- byte[] from = range[0];
- byte[] to = range[1];
- Path path = new Path(familyDir, "hfile_" + hfileIdx++);
- HFileTestUtil.createHFile(util.getConfiguration(), fs, path, fam, qual, from, to, factor);
- if (useMap) {
- last = path;
- list.add(path);
- }
- }
- int expectedRows = hfileIdx * factor;
-
- final TableName tableName = htd.getTableName();
- if (!util.getHBaseAdmin().tableExists(tableName) && (preCreateTable || map != null)) {
- util.getAdmin().createTable(htd, tableSplitKeys);
- }
-
- Configuration conf = util.getConfiguration();
- if (copyFiles) {
- conf.setBoolean(LoadIncrementalHFiles.ALWAYS_COPY_FILES, true);
- }
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
- String [] args= {dir.toString(), tableName.toString()};
- if (useMap) {
- if (deleteFile) fs.delete(last);
- Map<LoadQueueItem, ByteBuffer> loaded = loader.run(null, map, tableName);
- if (deleteFile) {
- expectedRows -= 1000;
- for (LoadQueueItem item : loaded.keySet()) {
- if (item.hfilePath.getName().equals(last.getName())) {
- fail(last + " should be missing");
- }
- }
- }
- } else {
- loader.run(args);
- }
-
- if (copyFiles) {
- for (Path p : list) {
- assertTrue(p + " should exist", fs.exists(p));
- }
- }
-
- Table table = util.getConnection().getTable(tableName);
- try {
- assertEquals(initRowCount + expectedRows, util.countRows(table));
- } finally {
- table.close();
- }
-
- return expectedRows;
- }
-
- private void runTest(String testName, HTableDescriptor htd, BloomType bloomType,
- boolean preCreateTable, byte[][] tableSplitKeys, byte[][][] hfileRanges, boolean useMap,
- boolean copyFiles) throws Exception {
- loadHFiles(testName, htd, util, FAMILY, QUALIFIER, preCreateTable, tableSplitKeys,
- hfileRanges, useMap, true, copyFiles, 0, 1000);
-
- final TableName tableName = htd.getTableName();
- // verify staging folder has been cleaned up
- Path stagingBasePath = new Path(FSUtils.getRootDir(util.getConfiguration()), HConstants.BULKLOAD_STAGING_DIR_NAME);
- FileSystem fs = util.getTestFileSystem();
- if(fs.exists(stagingBasePath)) {
- FileStatus[] files = fs.listStatus(stagingBasePath);
- for(FileStatus file : files) {
- assertTrue("Folder=" + file.getPath() + " is not cleaned up.",
- file.getPath().getName() != "DONOTERASE");
- }
- }
-
- util.deleteTable(tableName);
- }
-
- /**
- * Test that tags survive through a bulk load that needs to split hfiles.
- *
- * This test depends on the "hbase.client.rpc.codec" = KeyValueCodecWithTags so that the client
- * can get tags in the responses.
- */
- @Test(timeout = 60000)
- public void testTagsSurviveBulkLoadSplit() throws Exception {
- Path dir = util.getDataTestDirOnTestFS(tn.getMethodName());
- FileSystem fs = util.getTestFileSystem();
- dir = dir.makeQualified(fs);
- Path familyDir = new Path(dir, Bytes.toString(FAMILY));
- // table has these split points
- byte [][] tableSplitKeys = new byte[][] {
- Bytes.toBytes("aaa"), Bytes.toBytes("fff"), Bytes.toBytes("jjj"),
- Bytes.toBytes("ppp"), Bytes.toBytes("uuu"), Bytes.toBytes("zzz"),
- };
-
- // creating an hfile that has values that span the split points.
- byte[] from = Bytes.toBytes("ddd");
- byte[] to = Bytes.toBytes("ooo");
- HFileTestUtil.createHFileWithTags(util.getConfiguration(), fs,
- new Path(familyDir, tn.getMethodName()+"_hfile"),
- FAMILY, QUALIFIER, from, to, 1000);
- int expectedRows = 1000;
-
- TableName tableName = TableName.valueOf(tn.getMethodName());
- HTableDescriptor htd = buildHTD(tableName, BloomType.NONE);
- util.getAdmin().createTable(htd, tableSplitKeys);
-
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(util.getConfiguration());
- String [] args= {dir.toString(), tableName.toString()};
- loader.run(args);
-
- Table table = util.getConnection().getTable(tableName);
- try {
- assertEquals(expectedRows, util.countRows(table));
- HFileTestUtil.verifyTags(table);
- } finally {
- table.close();
- }
-
- util.deleteTable(tableName);
- }
-
- /**
- * Test loading into a column family that does not exist.
- */
- @Test(timeout = 60000)
- public void testNonexistentColumnFamilyLoad() throws Exception {
- String testName = tn.getMethodName();
- byte[][][] hFileRanges = new byte[][][] {
- new byte[][]{ Bytes.toBytes("aaa"), Bytes.toBytes("ccc") },
- new byte[][]{ Bytes.toBytes("ddd"), Bytes.toBytes("ooo") },
- };
-
- final byte[] TABLE = Bytes.toBytes("mytable_"+testName);
- HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(TABLE));
- // set real family name to upper case in purpose to simulate the case that
- // family name in HFiles is invalid
- HColumnDescriptor family =
- new HColumnDescriptor(Bytes.toBytes(new String(FAMILY).toUpperCase(Locale.ROOT)));
- htd.addFamily(family);
-
- try {
- runTest(testName, htd, BloomType.NONE, true, SPLIT_KEYS, hFileRanges, false, false);
- assertTrue("Loading into table with non-existent family should have failed", false);
- } catch (Exception e) {
- assertTrue("IOException expected", e instanceof IOException);
- // further check whether the exception message is correct
- String errMsg = e.getMessage();
- assertTrue("Incorrect exception message, expected message: ["
- + EXPECTED_MSG_FOR_NON_EXISTING_FAMILY + "], current message: [" + errMsg + "]",
- errMsg.contains(EXPECTED_MSG_FOR_NON_EXISTING_FAMILY));
- }
- }
-
- @Test(timeout = 120000)
- public void testNonHfileFolderWithUnmatchedFamilyName() throws Exception {
- testNonHfileFolder("testNonHfileFolderWithUnmatchedFamilyName", true);
- }
-
- @Test(timeout = 120000)
- public void testNonHfileFolder() throws Exception {
- testNonHfileFolder("testNonHfileFolder", false);
- }
-
- /**
- * Write a random data file and a non-file in a dir with a valid family name
- * but not part of the table families. we should we able to bulkload without
- * getting the unmatched family exception. HBASE-13037/HBASE-13227
- */
- private void testNonHfileFolder(String tableName, boolean preCreateTable) throws Exception {
- Path dir = util.getDataTestDirOnTestFS(tableName);
- FileSystem fs = util.getTestFileSystem();
- dir = dir.makeQualified(fs);
-
- Path familyDir = new Path(dir, Bytes.toString(FAMILY));
- HFileTestUtil.createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_0"),
- FAMILY, QUALIFIER, Bytes.toBytes("begin"), Bytes.toBytes("end"), 500);
- createRandomDataFile(fs, new Path(familyDir, "012356789"), 16 * 1024);
-
- final String NON_FAMILY_FOLDER = "_logs";
- Path nonFamilyDir = new Path(dir, NON_FAMILY_FOLDER);
- fs.mkdirs(nonFamilyDir);
- fs.mkdirs(new Path(nonFamilyDir, "non-file"));
- createRandomDataFile(fs, new Path(nonFamilyDir, "012356789"), 16 * 1024);
-
- Table table = null;
- try {
- if (preCreateTable) {
- table = util.createTable(TableName.valueOf(tableName), FAMILY);
- } else {
- table = util.getConnection().getTable(TableName.valueOf(tableName));
- }
-
- final String[] args = {dir.toString(), tableName};
- new LoadIncrementalHFiles(util.getConfiguration()).run(args);
- assertEquals(500, util.countRows(table));
- } finally {
- if (table != null) {
- table.close();
- }
- fs.delete(dir, true);
- }
- }
-
- private static void createRandomDataFile(FileSystem fs, Path path, int size)
- throws IOException {
- FSDataOutputStream stream = fs.create(path);
- try {
- byte[] data = new byte[1024];
- for (int i = 0; i < data.length; ++i) {
- data[i] = (byte)(i & 0xff);
- }
- while (size >= data.length) {
- stream.write(data, 0, data.length);
- size -= data.length;
- }
- if (size > 0) {
- stream.write(data, 0, size);
- }
- } finally {
- stream.close();
- }
- }
-
- @Test(timeout = 120000)
- public void testSplitStoreFile() throws IOException {
- Path dir = util.getDataTestDirOnTestFS("testSplitHFile");
- FileSystem fs = util.getTestFileSystem();
- Path testIn = new Path(dir, "testhfile");
- HColumnDescriptor familyDesc = new HColumnDescriptor(FAMILY);
- HFileTestUtil.createHFile(util.getConfiguration(), fs, testIn, FAMILY, QUALIFIER,
- Bytes.toBytes("aaa"), Bytes.toBytes("zzz"), 1000);
-
- Path bottomOut = new Path(dir, "bottom.out");
- Path topOut = new Path(dir, "top.out");
-
- LoadIncrementalHFiles.splitStoreFile(
- util.getConfiguration(), testIn,
- familyDesc, Bytes.toBytes("ggg"),
- bottomOut,
- topOut);
-
- int rowCount = verifyHFile(bottomOut);
- rowCount += verifyHFile(topOut);
- assertEquals(1000, rowCount);
- }
-
- @Test
- public void testSplitStoreFileWithNoneToNone() throws IOException {
- testSplitStoreFileWithDifferentEncoding(DataBlockEncoding.NONE, DataBlockEncoding.NONE);
- }
-
- @Test
- public void testSplitStoreFileWithEncodedToEncoded() throws IOException {
- testSplitStoreFileWithDifferentEncoding(DataBlockEncoding.DIFF, DataBlockEncoding.DIFF);
- }
-
- @Test
- public void testSplitStoreFileWithEncodedToNone() throws IOException {
- testSplitStoreFileWithDifferentEncoding(DataBlockEncoding.DIFF, DataBlockEncoding.NONE);
- }
-
- @Test
- public void testSplitStoreFileWithNoneToEncoded() throws IOException {
- testSplitStoreFileWithDifferentEncoding(DataBlockEncoding.NONE, DataBlockEncoding.DIFF);
- }
-
- private void testSplitStoreFileWithDifferentEncoding(DataBlockEncoding bulkloadEncoding,
- DataBlockEncoding cfEncoding) throws IOException {
- Path dir = util.getDataTestDirOnTestFS("testSplitHFileWithDifferentEncoding");
- FileSystem fs = util.getTestFileSystem();
- Path testIn = new Path(dir, "testhfile");
- HColumnDescriptor familyDesc = new HColumnDescriptor(FAMILY);
- familyDesc.setDataBlockEncoding(cfEncoding);
- HFileTestUtil.createHFileWithDataBlockEncoding(
- util.getConfiguration(), fs, testIn, bulkloadEncoding,
- FAMILY, QUALIFIER, Bytes.toBytes("aaa"), Bytes.toBytes("zzz"), 1000);
-
- Path bottomOut = new Path(dir, "bottom.out");
- Path topOut = new Path(dir, "top.out");
-
- LoadIncrementalHFiles.splitStoreFile(
- util.getConfiguration(), testIn,
- familyDesc, Bytes.toBytes("ggg"),
- bottomOut,
- topOut);
-
- int rowCount = verifyHFile(bottomOut);
- rowCount += verifyHFile(topOut);
- assertEquals(1000, rowCount);
- }
-
- private int verifyHFile(Path p) throws IOException {
- Configuration conf = util.getConfiguration();
- HFile.Reader reader =
- HFile.createReader(p.getFileSystem(conf), p, new CacheConfig(conf), true, conf);
- reader.loadFileInfo();
- HFileScanner scanner = reader.getScanner(false, false);
- scanner.seekTo();
- int count = 0;
- do {
- count++;
- } while (scanner.next());
- assertTrue(count > 0);
- reader.close();
- return count;
- }
-
- private void addStartEndKeysForTest(TreeMap<byte[], Integer> map, byte[] first, byte[] last) {
- Integer value = map.containsKey(first)?map.get(first):0;
- map.put(first, value+1);
-
- value = map.containsKey(last)?map.get(last):0;
- map.put(last, value-1);
- }
-
- @Test(timeout = 120000)
- public void testInferBoundaries() {
- TreeMap<byte[], Integer> map = new TreeMap<>(Bytes.BYTES_COMPARATOR);
-
- /* Toy example
- * c---------i o------p s---------t v------x
- * a------e g-----k m-------------q r----s u----w
- *
- * Should be inferred as:
- * a-----------------k m-------------q r--------------t u---------x
- *
- * The output should be (m,r,u)
- */
-
- String first;
- String last;
-
- first = "a"; last = "e";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "r"; last = "s";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "o"; last = "p";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "g"; last = "k";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "v"; last = "x";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "c"; last = "i";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "m"; last = "q";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "s"; last = "t";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- first = "u"; last = "w";
- addStartEndKeysForTest(map, first.getBytes(), last.getBytes());
-
- byte[][] keysArray = LoadIncrementalHFiles.inferBoundaries(map);
- byte[][] compare = new byte[3][];
- compare[0] = "m".getBytes();
- compare[1] = "r".getBytes();
- compare[2] = "u".getBytes();
-
- assertEquals(keysArray.length, 3);
-
- for (int row = 0; row<keysArray.length; row++){
- assertArrayEquals(keysArray[row], compare[row]);
- }
- }
-
- @Test(timeout = 60000)
- public void testLoadTooMayHFiles() throws Exception {
- Path dir = util.getDataTestDirOnTestFS("testLoadTooMayHFiles");
- FileSystem fs = util.getTestFileSystem();
- dir = dir.makeQualified(fs);
- Path familyDir = new Path(dir, Bytes.toString(FAMILY));
-
- byte[] from = Bytes.toBytes("begin");
- byte[] to = Bytes.toBytes("end");
- for (int i = 0; i <= MAX_FILES_PER_REGION_PER_FAMILY; i++) {
- HFileTestUtil.createHFile(util.getConfiguration(), fs, new Path(familyDir, "hfile_"
- + i), FAMILY, QUALIFIER, from, to, 1000);
- }
-
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(util.getConfiguration());
- String [] args= {dir.toString(), "mytable_testLoadTooMayHFiles"};
- try {
- loader.run(args);
- fail("Bulk loading too many files should fail");
- } catch (IOException ie) {
- assertTrue(ie.getMessage().contains("Trying to load more than "
- + MAX_FILES_PER_REGION_PER_FAMILY + " hfiles"));
- }
- }
-
- @Test(expected = TableNotFoundException.class)
- public void testWithoutAnExistingTableAndCreateTableSetToNo() throws Exception {
- Configuration conf = util.getConfiguration();
- conf.set(LoadIncrementalHFiles.CREATE_TABLE_CONF_KEY, "no");
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
- String[] args = { "directory", "nonExistingTable" };
- loader.run(args);
- }
-
- @Test(timeout = 120000)
- public void testTableWithCFNameStartWithUnderScore() throws Exception {
- Path dir = util.getDataTestDirOnTestFS("cfNameStartWithUnderScore");
- FileSystem fs = util.getTestFileSystem();
- dir = dir.makeQualified(fs.getUri(), fs.getWorkingDirectory());
- String family = "_cf";
- Path familyDir = new Path(dir, family);
-
- byte[] from = Bytes.toBytes("begin");
- byte[] to = Bytes.toBytes("end");
- Configuration conf = util.getConfiguration();
- String tableName = tn.getMethodName();
- Table table = util.createTable(TableName.valueOf(tableName), family);
- HFileTestUtil.createHFile(conf, fs, new Path(familyDir, "hfile"), Bytes.toBytes(family),
- QUALIFIER, from, to, 1000);
-
- LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
- String[] args = { dir.toString(), tableName };
- try {
- loader.run(args);
- assertEquals(1000, util.countRows(table));
- } finally {
- if (null != table) {
- table.close();
- }
- }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/hbase/blob/9e53f292/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestScannerWithBulkload.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestScannerWithBulkload.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestScannerWithBulkload.java
index 07dd2a9..5dce4ad 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestScannerWithBulkload.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestScannerWithBulkload.java
@@ -40,9 +40,9 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.hfile.HFile;
import org.apache.hadoop.hbase.io.hfile.HFileContext;
-import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.AfterClass;
import org.junit.Assert;