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.mapreduce;
019
020import static org.apache.hadoop.hbase.regionserver.HStoreFile.BULKLOAD_TASK_KEY;
021import static org.apache.hadoop.hbase.regionserver.HStoreFile.BULKLOAD_TIME_KEY;
022import static org.apache.hadoop.hbase.regionserver.HStoreFile.EXCLUDE_FROM_MINOR_COMPACTION_KEY;
023import static org.apache.hadoop.hbase.regionserver.HStoreFile.MAJOR_COMPACTION_KEY;
024
025import java.io.IOException;
026import java.io.UncheckedIOException;
027import java.io.UnsupportedEncodingException;
028import java.net.InetSocketAddress;
029import java.net.URLDecoder;
030import java.net.URLEncoder;
031import java.nio.charset.Charset;
032import java.util.ArrayList;
033import java.util.Arrays;
034import java.util.Collections;
035import java.util.List;
036import java.util.Map;
037import java.util.Map.Entry;
038import java.util.Set;
039import java.util.TreeMap;
040import java.util.TreeSet;
041import java.util.UUID;
042import java.util.function.Function;
043import java.util.stream.Collectors;
044import org.apache.commons.lang3.StringUtils;
045import org.apache.hadoop.conf.Configuration;
046import org.apache.hadoop.fs.FileSystem;
047import org.apache.hadoop.fs.Path;
048import org.apache.hadoop.hbase.Cell;
049import org.apache.hadoop.hbase.CellUtil;
050import org.apache.hadoop.hbase.ExtendedCell;
051import org.apache.hadoop.hbase.HConstants;
052import org.apache.hadoop.hbase.HRegionLocation;
053import org.apache.hadoop.hbase.KeyValue;
054import org.apache.hadoop.hbase.KeyValueUtil;
055import org.apache.hadoop.hbase.PrivateCellUtil;
056import org.apache.hadoop.hbase.TableName;
057import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
058import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
059import org.apache.hadoop.hbase.client.Connection;
060import org.apache.hadoop.hbase.client.ConnectionFactory;
061import org.apache.hadoop.hbase.client.Put;
062import org.apache.hadoop.hbase.client.RegionLocator;
063import org.apache.hadoop.hbase.client.Table;
064import org.apache.hadoop.hbase.client.TableDescriptor;
065import org.apache.hadoop.hbase.fs.HFileSystem;
066import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
067import org.apache.hadoop.hbase.io.compress.Compression;
068import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
069import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
070import org.apache.hadoop.hbase.io.hfile.CacheConfig;
071import org.apache.hadoop.hbase.io.hfile.HFile;
072import org.apache.hadoop.hbase.io.hfile.HFileContext;
073import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
074import org.apache.hadoop.hbase.io.hfile.HFileWriterImpl;
075import org.apache.hadoop.hbase.regionserver.BloomType;
076import org.apache.hadoop.hbase.regionserver.HStore;
077import org.apache.hadoop.hbase.regionserver.StoreFileWriter;
078import org.apache.hadoop.hbase.regionserver.StoreUtils;
079import org.apache.hadoop.hbase.util.BloomFilterUtil;
080import org.apache.hadoop.hbase.util.Bytes;
081import org.apache.hadoop.hbase.util.CommonFSUtils;
082import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
083import org.apache.hadoop.hbase.util.MapReduceExtendedCell;
084import org.apache.hadoop.io.NullWritable;
085import org.apache.hadoop.io.SequenceFile;
086import org.apache.hadoop.io.Text;
087import org.apache.hadoop.io.Writable;
088import org.apache.hadoop.mapreduce.Job;
089import org.apache.hadoop.mapreduce.OutputCommitter;
090import org.apache.hadoop.mapreduce.OutputFormat;
091import org.apache.hadoop.mapreduce.RecordWriter;
092import org.apache.hadoop.mapreduce.TaskAttemptContext;
093import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
094import org.apache.hadoop.mapreduce.lib.output.PathOutputCommitter;
095import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
096import org.apache.yetus.audience.InterfaceAudience;
097import org.slf4j.Logger;
098import org.slf4j.LoggerFactory;
099
100/**
101 * Writes HFiles. Passed Cells must arrive in order. Writes current time as the sequence id for the
102 * file. Sets the major compacted attribute on created {@link HFile}s. Calling write(null,null) will
103 * forcibly roll all HFiles being written.
104 * <p>
105 * Using this class as part of a MapReduce job is best done using
106 * {@link #configureIncrementalLoad(Job, TableDescriptor, RegionLocator)}.
107 */
108@InterfaceAudience.Public
109public class HFileOutputFormat2 extends FileOutputFormat<ImmutableBytesWritable, Cell> {
110  private static final Logger LOG = LoggerFactory.getLogger(HFileOutputFormat2.class);
111
112  static class TableInfo {
113    private TableDescriptor tableDesctiptor;
114    private RegionLocator regionLocator;
115
116    public TableInfo(TableDescriptor tableDesctiptor, RegionLocator regionLocator) {
117      this.tableDesctiptor = tableDesctiptor;
118      this.regionLocator = regionLocator;
119    }
120
121    public TableDescriptor getTableDescriptor() {
122      return tableDesctiptor;
123    }
124
125    public RegionLocator getRegionLocator() {
126      return regionLocator;
127    }
128  }
129
130  protected static final byte[] tableSeparator = Bytes.toBytes(";");
131
132  protected static byte[] combineTableNameSuffix(byte[] tableName, byte[] suffix) {
133    return Bytes.add(tableName, tableSeparator, suffix);
134  }
135
136  // The following constants are private since these are used by
137  // HFileOutputFormat2 to internally transfer data between job setup and
138  // reducer run using conf.
139  // These should not be changed by the client.
140  static final String COMPRESSION_FAMILIES_CONF_KEY =
141    "hbase.hfileoutputformat.families.compression";
142  static final String BLOOM_TYPE_FAMILIES_CONF_KEY = "hbase.hfileoutputformat.families.bloomtype";
143  static final String BLOOM_PARAM_FAMILIES_CONF_KEY = "hbase.hfileoutputformat.families.bloomparam";
144  static final String BLOCK_SIZE_FAMILIES_CONF_KEY = "hbase.mapreduce.hfileoutputformat.blocksize";
145  static final String DATABLOCK_ENCODING_FAMILIES_CONF_KEY =
146    "hbase.mapreduce.hfileoutputformat.families.datablock.encoding";
147
148  // This constant is public since the client can modify this when setting
149  // up their conf object and thus refer to this symbol.
150  // It is present for backwards compatibility reasons. Use it only to
151  // override the auto-detection of datablock encoding and compression.
152  public static final String DATABLOCK_ENCODING_OVERRIDE_CONF_KEY =
153    "hbase.mapreduce.hfileoutputformat.datablock.encoding";
154  public static final String COMPRESSION_OVERRIDE_CONF_KEY =
155    "hbase.mapreduce.hfileoutputformat.compression";
156
157  /**
158   * Keep locality while generating HFiles for bulkload. See HBASE-12596
159   */
160  public static final String LOCALITY_SENSITIVE_CONF_KEY =
161    "hbase.bulkload.locality.sensitive.enabled";
162  private static final boolean DEFAULT_LOCALITY_SENSITIVE = true;
163  static final String OUTPUT_TABLE_NAME_CONF_KEY = "hbase.mapreduce.hfileoutputformat.table.name";
164  public static final String MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY =
165    "hbase.mapreduce.use.multi.table.hfileoutputformat";
166  public static final boolean MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_DEFAULT = true;
167
168  /**
169   * ExtendedCell and ExtendedCellSerialization are InterfaceAudience.Private. We expose this config
170   * for internal usage in jobs like WALPlayer which need to use features of ExtendedCell.
171   */
172  @InterfaceAudience.Private
173  public static final String EXTENDED_CELL_SERIALIZATION_ENABLED_KEY =
174    "hbase.mapreduce.hfileoutputformat.extendedcell.enabled";
175  static final boolean EXTENDED_CELL_SERIALIZATION_ENABLED_DEFULT = false;
176
177  @InterfaceAudience.Private
178  public static final String DISK_BASED_SORTING_ENABLED_KEY =
179    "hbase.mapreduce.hfileoutputformat.disk.based.sorting.enabled";
180  private static final boolean DISK_BASED_SORTING_ENABLED_DEFAULT = false;
181
182  public static final String REMOTE_CLUSTER_CONF_PREFIX = "hbase.hfileoutputformat.remote.cluster.";
183  public static final String REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY =
184    REMOTE_CLUSTER_CONF_PREFIX + "zookeeper.quorum";
185  public static final String REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY =
186    REMOTE_CLUSTER_CONF_PREFIX + "zookeeper." + HConstants.CLIENT_PORT_STR;
187  public static final String REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY =
188    REMOTE_CLUSTER_CONF_PREFIX + HConstants.ZOOKEEPER_ZNODE_PARENT;
189
190  public static final String STORAGE_POLICY_PROPERTY = HStore.BLOCK_STORAGE_POLICY_KEY;
191  public static final String STORAGE_POLICY_PROPERTY_CF_PREFIX = STORAGE_POLICY_PROPERTY + ".";
192
193  @Override
194  public RecordWriter<ImmutableBytesWritable, Cell>
195    getRecordWriter(final TaskAttemptContext context) throws IOException, InterruptedException {
196    return createRecordWriter(context, this.getOutputCommitter(context));
197  }
198
199  protected static byte[] getTableNameSuffixedWithFamily(byte[] tableName, byte[] family) {
200    return combineTableNameSuffix(tableName, family);
201  }
202
203  static <V extends Cell> RecordWriter<ImmutableBytesWritable, V> createRecordWriter(
204    final TaskAttemptContext context, final OutputCommitter committer) throws IOException {
205
206    // Get the path of the temporary output file
207    final Path outputDir = ((PathOutputCommitter) committer).getWorkPath();
208    final Configuration conf = context.getConfiguration();
209    final boolean writeMultipleTables =
210      conf.getBoolean(MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY, false);
211    final String writeTableNames = conf.get(OUTPUT_TABLE_NAME_CONF_KEY);
212    if (writeTableNames == null || writeTableNames.isEmpty()) {
213      throw new IllegalArgumentException("" + OUTPUT_TABLE_NAME_CONF_KEY + " cannot be empty");
214    }
215    final FileSystem fs = outputDir.getFileSystem(conf);
216    // These configs. are from hbase-*.xml
217    final long maxsize =
218      conf.getLong(HConstants.HREGION_MAX_FILESIZE, HConstants.DEFAULT_MAX_FILE_SIZE);
219    // Invented config. Add to hbase-*.xml if other than default compression.
220    final String defaultCompressionStr =
221      conf.get("hfile.compression", Compression.Algorithm.NONE.getName());
222    final Algorithm defaultCompression = HFileWriterImpl.compressionByName(defaultCompressionStr);
223    String compressionStr = conf.get(COMPRESSION_OVERRIDE_CONF_KEY);
224    final Algorithm overriddenCompression =
225      compressionStr != null ? Compression.getCompressionAlgorithmByName(compressionStr) : null;
226    final boolean compactionExclude =
227      conf.getBoolean("hbase.mapreduce.hfileoutputformat.compaction.exclude", false);
228    final Set<String> allTableNames = Arrays
229      .stream(writeTableNames.split(Bytes.toString(tableSeparator))).collect(Collectors.toSet());
230
231    // create a map from column family to the compression algorithm
232    final Map<byte[], Algorithm> compressionMap = createFamilyCompressionMap(conf);
233    final Map<byte[], BloomType> bloomTypeMap = createFamilyBloomTypeMap(conf);
234    final Map<byte[], String> bloomParamMap = createFamilyBloomParamMap(conf);
235    final Map<byte[], Integer> blockSizeMap = createFamilyBlockSizeMap(conf);
236
237    String dataBlockEncodingStr = conf.get(DATABLOCK_ENCODING_OVERRIDE_CONF_KEY);
238    final Map<byte[], DataBlockEncoding> datablockEncodingMap =
239      createFamilyDataBlockEncodingMap(conf);
240    final DataBlockEncoding overriddenEncoding =
241      dataBlockEncodingStr != null ? DataBlockEncoding.valueOf(dataBlockEncodingStr) : null;
242
243    return new RecordWriter<ImmutableBytesWritable, V>() {
244      // Map of families to writers and how much has been output on the writer.
245      private final Map<byte[], WriterLength> writers = new TreeMap<>(Bytes.BYTES_COMPARATOR);
246      private final Map<byte[], byte[]> previousRows = new TreeMap<>(Bytes.BYTES_COMPARATOR);
247      private final long now = EnvironmentEdgeManager.currentTime();
248      private byte[] tableNameBytes = writeMultipleTables ? null : Bytes.toBytes(writeTableNames);
249
250      @Override
251      public void write(ImmutableBytesWritable row, V cell) throws IOException {
252        // null input == user explicitly wants to flush
253        if (row == null && cell == null) {
254          rollWriters(null);
255          return;
256        }
257
258        ExtendedCell kv = PrivateCellUtil.ensureExtendedCell(cell);
259        byte[] rowKey = CellUtil.cloneRow(kv);
260        int length = (PrivateCellUtil.estimatedSerializedSizeOf(kv)) - Bytes.SIZEOF_INT;
261        byte[] family = CellUtil.cloneFamily(kv);
262        if (writeMultipleTables) {
263          tableNameBytes = MultiTableHFileOutputFormat.getTableName(row.get());
264          tableNameBytes = TableName.valueOf(tableNameBytes).getNameWithNamespaceInclAsString()
265            .getBytes(Charset.defaultCharset());
266          if (!allTableNames.contains(Bytes.toString(tableNameBytes))) {
267            throw new IllegalArgumentException(
268              "TableName " + Bytes.toString(tableNameBytes) + " not expected");
269          }
270        }
271        byte[] tableAndFamily = getTableNameSuffixedWithFamily(tableNameBytes, family);
272
273        WriterLength wl = this.writers.get(tableAndFamily);
274
275        // If this is a new column family, verify that the directory exists
276        if (wl == null) {
277          Path writerPath = null;
278          if (writeMultipleTables) {
279            Path tableRelPath = getTableRelativePath(tableNameBytes);
280            writerPath = new Path(outputDir, new Path(tableRelPath, Bytes.toString(family)));
281          } else {
282            writerPath = new Path(outputDir, Bytes.toString(family));
283          }
284          fs.mkdirs(writerPath);
285          configureStoragePolicy(conf, fs, tableAndFamily, writerPath);
286        }
287
288        // This can only happen once a row is finished though
289        if (
290          wl != null && wl.written + length >= maxsize
291            && Bytes.compareTo(this.previousRows.get(family), rowKey) != 0
292        ) {
293          rollWriters(wl);
294        }
295
296        // create a new WAL writer, if necessary
297        if (wl == null || wl.writer == null) {
298          InetSocketAddress[] favoredNodes = null;
299          if (conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
300            HRegionLocation loc = null;
301            String tableName = Bytes.toString(tableNameBytes);
302            if (tableName != null) {
303              try (
304                Connection connection =
305                  ConnectionFactory.createConnection(createRemoteClusterConf(conf));
306                RegionLocator locator = connection.getRegionLocator(TableName.valueOf(tableName))) {
307                loc = locator.getRegionLocation(rowKey);
308              } catch (Throwable e) {
309                LOG.warn("Something wrong locating rowkey {} in {}", Bytes.toString(rowKey),
310                  tableName, e);
311                loc = null;
312              }
313            }
314            if (null == loc) {
315              LOG.trace("Failed get of location, use default writer {}", Bytes.toString(rowKey));
316            } else {
317              LOG.debug("First rowkey: [{}]", Bytes.toString(rowKey));
318              InetSocketAddress initialIsa =
319                new InetSocketAddress(loc.getHostname(), loc.getPort());
320              if (initialIsa.isUnresolved()) {
321                LOG.trace("Failed resolve address {}, use default writer", loc.getHostnamePort());
322              } else {
323                LOG.debug("Use favored nodes writer: {}", initialIsa.getHostString());
324                favoredNodes = new InetSocketAddress[] { initialIsa };
325              }
326            }
327          }
328          wl = getNewWriter(tableNameBytes, family, conf, favoredNodes);
329
330        }
331
332        // we now have the proper WAL writer. full steam ahead
333        PrivateCellUtil.updateLatestStamp(kv, this.now);
334        wl.writer.append((ExtendedCell) kv);
335        wl.written += length;
336
337        // Copy the row so we know when a row transition.
338        this.previousRows.put(family, rowKey);
339      }
340
341      private Path getTableRelativePath(byte[] tableNameBytes) {
342        String tableName = Bytes.toString(tableNameBytes);
343        String[] tableNameParts = tableName.split(":");
344        Path tableRelPath = new Path(tableNameParts[0]);
345        if (tableNameParts.length > 1) {
346          tableRelPath = new Path(tableRelPath, tableNameParts[1]);
347        }
348        return tableRelPath;
349      }
350
351      private void rollWriters(WriterLength writerLength) throws IOException {
352        if (writerLength != null) {
353          closeWriter(writerLength);
354        } else {
355          for (WriterLength wl : this.writers.values()) {
356            closeWriter(wl);
357          }
358        }
359      }
360
361      private void closeWriter(WriterLength wl) throws IOException {
362        if (wl.writer != null) {
363          LOG.info(
364            "Writer=" + wl.writer.getPath() + ((wl.written == 0) ? "" : ", wrote=" + wl.written));
365          close(wl.writer);
366          wl.writer = null;
367        }
368        wl.written = 0;
369      }
370
371      private Configuration createRemoteClusterConf(Configuration conf) {
372        final Configuration newConf = new Configuration(conf);
373
374        final String quorum = conf.get(REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY);
375        final String clientPort = conf.get(REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY);
376        final String parent = conf.get(REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY);
377
378        if (quorum != null && clientPort != null && parent != null) {
379          newConf.set(HConstants.ZOOKEEPER_QUORUM, quorum);
380          newConf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, Integer.parseInt(clientPort));
381          newConf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, parent);
382        }
383
384        for (Entry<String, String> entry : conf) {
385          String key = entry.getKey();
386          if (
387            REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY.equals(key)
388              || REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY.equals(key)
389              || REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY.equals(key)
390          ) {
391            // Handled them above
392            continue;
393          }
394
395          if (entry.getKey().startsWith(REMOTE_CLUSTER_CONF_PREFIX)) {
396            String originalKey = entry.getKey().substring(REMOTE_CLUSTER_CONF_PREFIX.length());
397            if (!originalKey.isEmpty()) {
398              newConf.set(originalKey, entry.getValue());
399            }
400          }
401        }
402
403        return newConf;
404      }
405
406      /*
407       * Create a new StoreFile.Writer.
408       * @return A WriterLength, containing a new StoreFile.Writer.
409       */
410      @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "BX_UNBOXING_IMMEDIATELY_REBOXED",
411          justification = "Not important")
412      private WriterLength getNewWriter(byte[] tableName, byte[] family, Configuration conf,
413        InetSocketAddress[] favoredNodes) throws IOException {
414        byte[] tableAndFamily = getTableNameSuffixedWithFamily(tableName, family);
415        Path familydir = new Path(outputDir, Bytes.toString(family));
416        if (writeMultipleTables) {
417          familydir =
418            new Path(outputDir, new Path(getTableRelativePath(tableName), Bytes.toString(family)));
419        }
420        WriterLength wl = new WriterLength();
421        Algorithm compression = overriddenCompression;
422        compression = compression == null ? compressionMap.get(tableAndFamily) : compression;
423        compression = compression == null ? defaultCompression : compression;
424        BloomType bloomType = bloomTypeMap.get(tableAndFamily);
425        bloomType = bloomType == null ? BloomType.NONE : bloomType;
426        String bloomParam = bloomParamMap.get(tableAndFamily);
427        if (bloomType == BloomType.ROWPREFIX_FIXED_LENGTH) {
428          conf.set(BloomFilterUtil.PREFIX_LENGTH_KEY, bloomParam);
429        }
430        Integer blockSize = blockSizeMap.get(tableAndFamily);
431        blockSize = blockSize == null ? HConstants.DEFAULT_BLOCKSIZE : blockSize;
432        DataBlockEncoding encoding = overriddenEncoding;
433        encoding = encoding == null ? datablockEncodingMap.get(tableAndFamily) : encoding;
434        encoding = encoding == null ? DataBlockEncoding.NONE : encoding;
435        HFileContextBuilder contextBuilder = new HFileContextBuilder().withCompression(compression)
436          .withDataBlockEncoding(encoding).withChecksumType(StoreUtils.getChecksumType(conf))
437          .withBytesPerCheckSum(StoreUtils.getBytesPerChecksum(conf)).withBlockSize(blockSize)
438          .withColumnFamily(family).withTableName(tableName)
439          .withCreateTime(EnvironmentEdgeManager.currentTime());
440
441        if (HFile.getFormatVersion(conf) >= HFile.MIN_FORMAT_VERSION_WITH_TAGS) {
442          contextBuilder.withIncludesTags(true);
443        }
444
445        HFileContext hFileContext = contextBuilder.build();
446        if (null == favoredNodes) {
447          wl.writer =
448            new StoreFileWriter.Builder(conf, CacheConfig.DISABLED, fs).withOutputDir(familydir)
449              .withBloomType(bloomType).withFileContext(hFileContext).build();
450        } else {
451          wl.writer = new StoreFileWriter.Builder(conf, CacheConfig.DISABLED, new HFileSystem(fs))
452            .withOutputDir(familydir).withBloomType(bloomType).withFileContext(hFileContext)
453            .withFavoredNodes(favoredNodes).build();
454        }
455
456        this.writers.put(tableAndFamily, wl);
457        return wl;
458      }
459
460      private void close(final StoreFileWriter w) throws IOException {
461        if (w != null) {
462          w.appendFileInfo(BULKLOAD_TIME_KEY, Bytes.toBytes(EnvironmentEdgeManager.currentTime()));
463          w.appendFileInfo(BULKLOAD_TASK_KEY, Bytes.toBytes(context.getTaskAttemptID().toString()));
464          w.appendFileInfo(MAJOR_COMPACTION_KEY, Bytes.toBytes(true));
465          w.appendFileInfo(EXCLUDE_FROM_MINOR_COMPACTION_KEY, Bytes.toBytes(compactionExclude));
466          w.appendTrackedTimestampsToMetadata();
467          w.close();
468        }
469      }
470
471      @Override
472      public void close(TaskAttemptContext c) throws IOException, InterruptedException {
473        for (WriterLength wl : this.writers.values()) {
474          close(wl.writer);
475        }
476      }
477    };
478  }
479
480  /**
481   * Configure block storage policy for CF after the directory is created.
482   */
483  static void configureStoragePolicy(final Configuration conf, final FileSystem fs,
484    byte[] tableAndFamily, Path cfPath) {
485    if (null == conf || null == fs || null == tableAndFamily || null == cfPath) {
486      return;
487    }
488
489    String policy = conf.get(STORAGE_POLICY_PROPERTY_CF_PREFIX + Bytes.toString(tableAndFamily),
490      conf.get(STORAGE_POLICY_PROPERTY));
491    CommonFSUtils.setStoragePolicy(fs, cfPath, policy);
492  }
493
494  /*
495   * Data structure to hold a Writer and amount of data written on it.
496   */
497  static class WriterLength {
498    long written = 0;
499    StoreFileWriter writer = null;
500  }
501
502  /**
503   * Return the start keys of all of the regions in this table, as a list of ImmutableBytesWritable.
504   */
505  private static List<ImmutableBytesWritable> getRegionStartKeys(List<RegionLocator> regionLocators,
506    boolean writeMultipleTables) throws IOException {
507
508    ArrayList<ImmutableBytesWritable> ret = new ArrayList<>();
509    for (RegionLocator regionLocator : regionLocators) {
510      TableName tableName = regionLocator.getName();
511      LOG.info("Looking up current regions for table " + tableName);
512      byte[][] byteKeys = regionLocator.getStartKeys();
513      for (byte[] byteKey : byteKeys) {
514        byte[] fullKey = byteKey; // HFileOutputFormat2 use case
515        if (writeMultipleTables) {
516          // MultiTableHFileOutputFormat use case
517          fullKey = combineTableNameSuffix(tableName.getName(), byteKey);
518        }
519        if (LOG.isDebugEnabled()) {
520          LOG.debug("SplitPoint startkey for " + tableName + ": " + Bytes.toStringBinary(fullKey));
521        }
522        ret.add(new ImmutableBytesWritable(fullKey));
523      }
524    }
525    return ret;
526  }
527
528  /**
529   * Write out a {@link SequenceFile} that can be read by {@link TotalOrderPartitioner} that
530   * contains the split points in startKeys.
531   */
532  @SuppressWarnings("deprecation")
533  private static void writePartitions(Configuration conf, Path partitionsPath,
534    List<ImmutableBytesWritable> startKeys, boolean writeMultipleTables) throws IOException {
535    LOG.info("Writing partition information to " + partitionsPath);
536    if (startKeys.isEmpty()) {
537      throw new IllegalArgumentException("No regions passed");
538    }
539
540    // We're generating a list of split points, and we don't ever
541    // have keys < the first region (which has an empty start key)
542    // so we need to remove it. Otherwise we would end up with an
543    // empty reducer with index 0
544    TreeSet<ImmutableBytesWritable> sorted = new TreeSet<>(startKeys);
545    ImmutableBytesWritable first = sorted.first();
546    if (writeMultipleTables) {
547      first =
548        new ImmutableBytesWritable(MultiTableHFileOutputFormat.getSuffix(sorted.first().get()));
549    }
550    if (!first.equals(HConstants.EMPTY_BYTE_ARRAY)) {
551      throw new IllegalArgumentException(
552        "First region of table should have empty start key. Instead has: "
553          + Bytes.toStringBinary(first.get()));
554    }
555    sorted.remove(sorted.first());
556
557    // Write the actual file
558    FileSystem fs = partitionsPath.getFileSystem(conf);
559    boolean diskBasedSortingEnabled = diskBasedSortingEnabled(conf);
560    Class<? extends Writable> keyClass =
561      diskBasedSortingEnabled ? KeyOnlyCellComparable.class : ImmutableBytesWritable.class;
562    SequenceFile.Writer writer =
563      SequenceFile.createWriter(fs, conf, partitionsPath, keyClass, NullWritable.class);
564
565    try {
566      for (ImmutableBytesWritable startKey : sorted) {
567        Writable writable = diskBasedSortingEnabled
568          ? new KeyOnlyCellComparable(KeyValueUtil.createFirstOnRow(startKey.get()))
569          : startKey;
570
571        writer.append(writable, NullWritable.get());
572      }
573    } finally {
574      writer.close();
575    }
576  }
577
578  /**
579   * Configure a MapReduce Job to perform an incremental load into the given table. This
580   * <ul>
581   * <li>Inspects the table to configure a total order partitioner</li>
582   * <li>Uploads the partitions file to the cluster and adds it to the DistributedCache</li>
583   * <li>Sets the number of reduce tasks to match the current number of regions</li>
584   * <li>Sets the output key/value class to match HFileOutputFormat2's requirements</li>
585   * <li>Sets the reducer up to perform the appropriate sorting (either KeyValueSortReducer or
586   * PutSortReducer)</li>
587   * <li>Sets the HBase cluster key to load region locations for locality-sensitive</li>
588   * </ul>
589   * The user should be sure to set the map output value class to either KeyValue or Put before
590   * running this function.
591   */
592  public static void configureIncrementalLoad(Job job, Table table, RegionLocator regionLocator)
593    throws IOException {
594    configureIncrementalLoad(job, table.getDescriptor(), regionLocator);
595    configureForRemoteCluster(job, table.getConfiguration());
596  }
597
598  /**
599   * Configure a MapReduce Job to perform an incremental load into the given table. This
600   * <ul>
601   * <li>Inspects the table to configure a total order partitioner</li>
602   * <li>Uploads the partitions file to the cluster and adds it to the DistributedCache</li>
603   * <li>Sets the number of reduce tasks to match the current number of regions</li>
604   * <li>Sets the output key/value class to match HFileOutputFormat2's requirements</li>
605   * <li>Sets the reducer up to perform the appropriate sorting (either KeyValueSortReducer or
606   * PutSortReducer)</li>
607   * </ul>
608   * The user should be sure to set the map output value class to either KeyValue or Put before
609   * running this function.
610   */
611  public static void configureIncrementalLoad(Job job, TableDescriptor tableDescriptor,
612    RegionLocator regionLocator) throws IOException {
613    ArrayList<TableInfo> singleTableInfo = new ArrayList<>();
614    singleTableInfo.add(new TableInfo(tableDescriptor, regionLocator));
615    configureIncrementalLoad(job, singleTableInfo, HFileOutputFormat2.class);
616  }
617
618  public static boolean diskBasedSortingEnabled(Configuration conf) {
619    return conf.getBoolean(DISK_BASED_SORTING_ENABLED_KEY, DISK_BASED_SORTING_ENABLED_DEFAULT);
620  }
621
622  static void configureIncrementalLoad(Job job, List<TableInfo> multiTableInfo,
623    Class<? extends OutputFormat<?, ?>> cls) throws IOException {
624    Configuration conf = job.getConfiguration();
625    job.setOutputKeyClass(ImmutableBytesWritable.class);
626    job.setOutputValueClass(MapReduceExtendedCell.class);
627    job.setOutputFormatClass(cls);
628
629    if (multiTableInfo.stream().distinct().count() != multiTableInfo.size()) {
630      throw new IllegalArgumentException("Duplicate entries found in TableInfo argument");
631    }
632    boolean writeMultipleTables = false;
633    if (MultiTableHFileOutputFormat.class.equals(cls)) {
634      writeMultipleTables = true;
635      conf.setBoolean(MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY, true);
636    }
637    // Based on the configured map output class, set the correct reducer to properly
638    // sort the incoming values.
639    // TODO it would be nice to pick one or the other of these formats.
640    boolean diskBasedSorting = diskBasedSortingEnabled(conf);
641
642    if (diskBasedSorting) {
643      job.setMapOutputKeyClass(KeyOnlyCellComparable.class);
644      job.setSortComparatorClass(KeyOnlyCellComparable.KeyOnlyCellComparator.class);
645      job.setReducerClass(PreSortedCellsReducer.class);
646    } else if (
647      KeyValue.class.equals(job.getMapOutputValueClass())
648        || MapReduceExtendedCell.class.equals(job.getMapOutputValueClass())
649    ) {
650      job.setReducerClass(CellSortReducer.class);
651    } else if (Put.class.equals(job.getMapOutputValueClass())) {
652      job.setReducerClass(PutSortReducer.class);
653    } else if (Text.class.equals(job.getMapOutputValueClass())) {
654      job.setReducerClass(TextSortReducer.class);
655    } else {
656      LOG.warn("Unknown map output value type:" + job.getMapOutputValueClass());
657    }
658
659    mergeSerializations(conf);
660
661    if (conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
662      LOG.info("bulkload locality sensitive enabled");
663    }
664
665    /* Now get the region start keys for every table required */
666    List<String> allTableNames = new ArrayList<>(multiTableInfo.size());
667    List<RegionLocator> regionLocators = new ArrayList<>(multiTableInfo.size());
668    List<TableDescriptor> tableDescriptors = new ArrayList<>(multiTableInfo.size());
669
670    for (TableInfo tableInfo : multiTableInfo) {
671      regionLocators.add(tableInfo.getRegionLocator());
672      String tn = writeMultipleTables
673        ? tableInfo.getRegionLocator().getName().getNameWithNamespaceInclAsString()
674        : tableInfo.getRegionLocator().getName().getNameAsString();
675      allTableNames.add(tn);
676      tableDescriptors.add(tableInfo.getTableDescriptor());
677    }
678    // Record tablenames for creating writer by favored nodes, and decoding compression,
679    // block size and other attributes of columnfamily per table
680    conf.set(OUTPUT_TABLE_NAME_CONF_KEY,
681      StringUtils.join(allTableNames, Bytes.toString(tableSeparator)));
682    List<ImmutableBytesWritable> startKeys =
683      getRegionStartKeys(regionLocators, writeMultipleTables);
684    // Use table's region boundaries for TOP split points.
685    LOG.info("Configuring " + startKeys.size() + " reduce partitions "
686      + "to match current region count for all tables");
687    job.setNumReduceTasks(startKeys.size());
688
689    configurePartitioner(job, startKeys, writeMultipleTables);
690    // Set compression algorithms based on column families
691
692    conf.set(COMPRESSION_FAMILIES_CONF_KEY,
693      serializeColumnFamilyAttribute(compressionDetails, tableDescriptors));
694    conf.set(BLOCK_SIZE_FAMILIES_CONF_KEY,
695      serializeColumnFamilyAttribute(blockSizeDetails, tableDescriptors));
696    conf.set(BLOOM_TYPE_FAMILIES_CONF_KEY,
697      serializeColumnFamilyAttribute(bloomTypeDetails, tableDescriptors));
698    conf.set(BLOOM_PARAM_FAMILIES_CONF_KEY,
699      serializeColumnFamilyAttribute(bloomParamDetails, tableDescriptors));
700    conf.set(DATABLOCK_ENCODING_FAMILIES_CONF_KEY,
701      serializeColumnFamilyAttribute(dataBlockEncodingDetails, tableDescriptors));
702
703    TableMapReduceUtil.addDependencyJars(job);
704    TableMapReduceUtil.initCredentials(job);
705    LOG.info("Incremental output configured for tables: " + StringUtils.join(allTableNames, ","));
706  }
707
708  private static void mergeSerializations(Configuration conf) {
709    List<String> serializations = new ArrayList<>();
710
711    // add any existing values that have been set
712    String[] existing = conf.getStrings("io.serializations");
713    if (existing != null) {
714      Collections.addAll(serializations, existing);
715    }
716
717    serializations.add(MutationSerialization.class.getName());
718    serializations.add(ResultSerialization.class.getName());
719
720    // Add ExtendedCellSerialization, if configured. Order matters here. Hadoop's
721    // SerializationFactory runs through serializations in the order they are registered.
722    // We want to register ExtendedCellSerialization before CellSerialization because both
723    // work for ExtendedCells but only ExtendedCellSerialization handles them properly.
724    if (
725      conf.getBoolean(EXTENDED_CELL_SERIALIZATION_ENABLED_KEY,
726        EXTENDED_CELL_SERIALIZATION_ENABLED_DEFULT)
727    ) {
728      serializations.add(ExtendedCellSerialization.class.getName());
729    }
730    serializations.add(CellSerialization.class.getName());
731
732    conf.setStrings("io.serializations", serializations.toArray(new String[0]));
733  }
734
735  public static void configureIncrementalLoadMap(Job job, TableDescriptor tableDescriptor)
736    throws IOException {
737    Configuration conf = job.getConfiguration();
738
739    job.setOutputKeyClass(ImmutableBytesWritable.class);
740    job.setOutputValueClass(MapReduceExtendedCell.class);
741    job.setOutputFormatClass(HFileOutputFormat2.class);
742
743    ArrayList<TableDescriptor> singleTableDescriptor = new ArrayList<>(1);
744    singleTableDescriptor.add(tableDescriptor);
745
746    conf.set(OUTPUT_TABLE_NAME_CONF_KEY, tableDescriptor.getTableName().getNameAsString());
747    // Set compression algorithms based on column families
748    conf.set(COMPRESSION_FAMILIES_CONF_KEY,
749      serializeColumnFamilyAttribute(compressionDetails, singleTableDescriptor));
750    conf.set(BLOCK_SIZE_FAMILIES_CONF_KEY,
751      serializeColumnFamilyAttribute(blockSizeDetails, singleTableDescriptor));
752    conf.set(BLOOM_TYPE_FAMILIES_CONF_KEY,
753      serializeColumnFamilyAttribute(bloomTypeDetails, singleTableDescriptor));
754    conf.set(BLOOM_PARAM_FAMILIES_CONF_KEY,
755      serializeColumnFamilyAttribute(bloomParamDetails, singleTableDescriptor));
756    conf.set(DATABLOCK_ENCODING_FAMILIES_CONF_KEY,
757      serializeColumnFamilyAttribute(dataBlockEncodingDetails, singleTableDescriptor));
758
759    TableMapReduceUtil.addDependencyJars(job);
760    TableMapReduceUtil.initCredentials(job);
761    LOG.info("Incremental table " + tableDescriptor.getTableName() + " output configured.");
762  }
763
764  /**
765   * Configure HBase cluster key for remote cluster to load region location for locality-sensitive
766   * if it's enabled. It's not necessary to call this method explicitly when the cluster key for
767   * HBase cluster to be used to load region location is configured in the job configuration. Call
768   * this method when another HBase cluster key is configured in the job configuration. For example,
769   * you should call when you load data from HBase cluster A using {@link TableInputFormat} and
770   * generate hfiles for HBase cluster B. Otherwise, HFileOutputFormat2 fetch location from cluster
771   * A and locality-sensitive won't working correctly.
772   * {@link #configureIncrementalLoad(Job, Table, RegionLocator)} calls this method using
773   * {@link Table#getConfiguration} as clusterConf. See HBASE-25608.
774   * @param job         which has configuration to be updated
775   * @param clusterConf which contains cluster key of the HBase cluster to be locality-sensitive
776   * @see #configureIncrementalLoad(Job, Table, RegionLocator)
777   * @see #LOCALITY_SENSITIVE_CONF_KEY
778   * @see #REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY
779   * @see #REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY
780   * @see #REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY
781   * @deprecated As of release 2.6.4, this will be removed in HBase 4.0.0 Use
782   *             {@link #configureForRemoteCluster(Job, Configuration)} instead.
783   */
784  @Deprecated
785  public static void configureRemoteCluster(Job job, Configuration clusterConf) {
786    try {
787      configureForRemoteCluster(job, clusterConf);
788    } catch (IOException e) {
789      LOG.error("Configure remote cluster error.", e);
790      throw new UncheckedIOException("Configure remote cluster error.", e);
791    }
792  }
793
794  /**
795   * Configure HBase cluster key for remote cluster to load region location for locality-sensitive
796   * if it's enabled. It's not necessary to call this method explicitly when the cluster key for
797   * HBase cluster to be used to load region location is configured in the job configuration. Call
798   * this method when another HBase cluster key is configured in the job configuration. For example,
799   * you should call when you load data from HBase cluster A using {@link TableInputFormat} and
800   * generate hfiles for HBase cluster B. Otherwise, HFileOutputFormat2 fetch location from cluster
801   * A and locality-sensitive won't working correctly. If authentication is enabled, it obtains the
802   * token for the specific cluster.
803   * @param job         which has configuration to be updated
804   * @param clusterConf which contains cluster key of the HBase cluster to be locality-sensitive
805   * @throws IOException Exception while initializing cluster credentials
806   */
807  public static void configureForRemoteCluster(Job job, Configuration clusterConf)
808    throws IOException {
809    Configuration conf = job.getConfiguration();
810
811    if (!conf.getBoolean(LOCALITY_SENSITIVE_CONF_KEY, DEFAULT_LOCALITY_SENSITIVE)) {
812      return;
813    }
814
815    final String quorum = clusterConf.get(HConstants.ZOOKEEPER_QUORUM);
816    final int clientPort = clusterConf.getInt(HConstants.ZOOKEEPER_CLIENT_PORT,
817      HConstants.DEFAULT_ZOOKEEPER_CLIENT_PORT);
818    final String parent =
819      clusterConf.get(HConstants.ZOOKEEPER_ZNODE_PARENT, HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
820
821    conf.set(REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY, quorum);
822    conf.setInt(REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY, clientPort);
823    conf.set(REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY, parent);
824
825    TableMapReduceUtil.initCredentialsForCluster(job, clusterConf);
826
827    LOG.info("ZK configs for remote cluster of bulkload is configured: " + quorum + ":" + clientPort
828      + "/" + parent);
829  }
830
831  /**
832   * Runs inside the task to deserialize column family to compression algorithm map from the
833   * configuration.
834   * @param conf to read the serialized values from
835   * @return a map from column family to the configured compression algorithm
836   */
837  @InterfaceAudience.Private
838  static Map<byte[], Algorithm> createFamilyCompressionMap(Configuration conf) {
839    Map<byte[], String> stringMap = createFamilyConfValueMap(conf, COMPRESSION_FAMILIES_CONF_KEY);
840    Map<byte[], Algorithm> compressionMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
841    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
842      Algorithm algorithm = HFileWriterImpl.compressionByName(e.getValue());
843      compressionMap.put(e.getKey(), algorithm);
844    }
845    return compressionMap;
846  }
847
848  /**
849   * Runs inside the task to deserialize column family to bloom filter type map from the
850   * configuration.
851   * @param conf to read the serialized values from
852   * @return a map from column family to the the configured bloom filter type
853   */
854  @InterfaceAudience.Private
855  static Map<byte[], BloomType> createFamilyBloomTypeMap(Configuration conf) {
856    Map<byte[], String> stringMap = createFamilyConfValueMap(conf, BLOOM_TYPE_FAMILIES_CONF_KEY);
857    Map<byte[], BloomType> bloomTypeMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
858    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
859      BloomType bloomType = BloomType.valueOf(e.getValue());
860      bloomTypeMap.put(e.getKey(), bloomType);
861    }
862    return bloomTypeMap;
863  }
864
865  /**
866   * Runs inside the task to deserialize column family to bloom filter param map from the
867   * configuration.
868   * @param conf to read the serialized values from
869   * @return a map from column family to the the configured bloom filter param
870   */
871  @InterfaceAudience.Private
872  static Map<byte[], String> createFamilyBloomParamMap(Configuration conf) {
873    return createFamilyConfValueMap(conf, BLOOM_PARAM_FAMILIES_CONF_KEY);
874  }
875
876  /**
877   * Runs inside the task to deserialize column family to block size map from the configuration.
878   * @param conf to read the serialized values from
879   * @return a map from column family to the configured block size
880   */
881  @InterfaceAudience.Private
882  static Map<byte[], Integer> createFamilyBlockSizeMap(Configuration conf) {
883    Map<byte[], String> stringMap = createFamilyConfValueMap(conf, BLOCK_SIZE_FAMILIES_CONF_KEY);
884    Map<byte[], Integer> blockSizeMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
885    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
886      Integer blockSize = Integer.parseInt(e.getValue());
887      blockSizeMap.put(e.getKey(), blockSize);
888    }
889    return blockSizeMap;
890  }
891
892  /**
893   * Runs inside the task to deserialize column family to data block encoding type map from the
894   * configuration.
895   * @param conf to read the serialized values from
896   * @return a map from column family to HFileDataBlockEncoder for the configured data block type
897   *         for the family
898   */
899  @InterfaceAudience.Private
900  static Map<byte[], DataBlockEncoding> createFamilyDataBlockEncodingMap(Configuration conf) {
901    Map<byte[], String> stringMap =
902      createFamilyConfValueMap(conf, DATABLOCK_ENCODING_FAMILIES_CONF_KEY);
903    Map<byte[], DataBlockEncoding> encoderMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
904    for (Map.Entry<byte[], String> e : stringMap.entrySet()) {
905      encoderMap.put(e.getKey(), DataBlockEncoding.valueOf((e.getValue())));
906    }
907    return encoderMap;
908  }
909
910  /**
911   * Run inside the task to deserialize column family to given conf value map.
912   * @param conf     to read the serialized values from
913   * @param confName conf key to read from the configuration
914   * @return a map of column family to the given configuration value
915   */
916  private static Map<byte[], String> createFamilyConfValueMap(Configuration conf, String confName) {
917    Map<byte[], String> confValMap = new TreeMap<>(Bytes.BYTES_COMPARATOR);
918    String confVal = conf.get(confName, "");
919    for (String familyConf : confVal.split("&")) {
920      String[] familySplit = familyConf.split("=");
921      if (familySplit.length != 2) {
922        continue;
923      }
924      try {
925        confValMap.put(Bytes.toBytes(URLDecoder.decode(familySplit[0], "UTF-8")),
926          URLDecoder.decode(familySplit[1], "UTF-8"));
927      } catch (UnsupportedEncodingException e) {
928        // will not happen with UTF-8 encoding
929        throw new AssertionError(e);
930      }
931    }
932    return confValMap;
933  }
934
935  /**
936   * Configure <code>job</code> with a TotalOrderPartitioner, partitioning against
937   * <code>splitPoints</code>. Cleans up the partitions file after job exists.
938   */
939  static void configurePartitioner(Job job, List<ImmutableBytesWritable> splitPoints,
940    boolean writeMultipleTables) throws IOException {
941    Configuration conf = job.getConfiguration();
942    // create the partitions file
943    FileSystem fs = FileSystem.get(conf);
944    String hbaseTmpFsDir =
945      conf.get(HConstants.TEMPORARY_FS_DIRECTORY_KEY, fs.getHomeDirectory() + "/hbase-staging");
946    Path partitionsPath =
947      fs.makeQualified(new Path(hbaseTmpFsDir, "partitions_" + UUID.randomUUID()));
948    writePartitions(conf, partitionsPath, splitPoints, writeMultipleTables);
949    fs.deleteOnExit(partitionsPath);
950
951    // configure job to use it
952    job.setPartitionerClass(TotalOrderPartitioner.class);
953    TotalOrderPartitioner.setPartitionFile(conf, partitionsPath);
954  }
955
956  @edu.umd.cs.findbugs.annotations.SuppressWarnings(
957      value = "RCN_REDUNDANT_NULLCHECK_OF_NONNULL_VALUE")
958  @InterfaceAudience.Private
959  static String serializeColumnFamilyAttribute(Function<ColumnFamilyDescriptor, String> fn,
960    List<TableDescriptor> allTables) throws UnsupportedEncodingException {
961    StringBuilder attributeValue = new StringBuilder();
962    int i = 0;
963    for (TableDescriptor tableDescriptor : allTables) {
964      if (tableDescriptor == null) {
965        // could happen with mock table instance
966        // CODEREVIEW: Can I set an empty string in conf if mock table instance?
967        return "";
968      }
969      for (ColumnFamilyDescriptor familyDescriptor : tableDescriptor.getColumnFamilies()) {
970        if (i++ > 0) {
971          attributeValue.append('&');
972        }
973        attributeValue.append(URLEncoder
974          .encode(Bytes.toString(combineTableNameSuffix(tableDescriptor.getTableName().getName(),
975            familyDescriptor.getName())), "UTF-8"));
976        attributeValue.append('=');
977        attributeValue.append(URLEncoder.encode(fn.apply(familyDescriptor), "UTF-8"));
978      }
979    }
980    // Get rid of the last ampersand
981    return attributeValue.toString();
982  }
983
984  /**
985   * Serialize column family to compression algorithm map to configuration. Invoked while
986   * configuring the MR job for incremental load.
987   */
988  @InterfaceAudience.Private
989  static Function<ColumnFamilyDescriptor, String> compressionDetails =
990    familyDescriptor -> familyDescriptor.getCompressionType().getName();
991
992  /**
993   * Serialize column family to block size map to configuration. Invoked while configuring the MR
994   * job for incremental load.
995   */
996  @InterfaceAudience.Private
997  static Function<ColumnFamilyDescriptor, String> blockSizeDetails =
998    familyDescriptor -> String.valueOf(familyDescriptor.getBlocksize());
999
1000  /**
1001   * Serialize column family to bloom type map to configuration. Invoked while configuring the MR
1002   * job for incremental load.
1003   */
1004  @InterfaceAudience.Private
1005  static Function<ColumnFamilyDescriptor, String> bloomTypeDetails = familyDescriptor -> {
1006    String bloomType = familyDescriptor.getBloomFilterType().toString();
1007    if (bloomType == null) {
1008      bloomType = ColumnFamilyDescriptorBuilder.DEFAULT_BLOOMFILTER.name();
1009    }
1010    return bloomType;
1011  };
1012
1013  /**
1014   * Serialize column family to bloom param map to configuration. Invoked while configuring the MR
1015   * job for incremental load.
1016   */
1017  @InterfaceAudience.Private
1018  static Function<ColumnFamilyDescriptor, String> bloomParamDetails = familyDescriptor -> {
1019    BloomType bloomType = familyDescriptor.getBloomFilterType();
1020    String bloomParam = "";
1021    if (bloomType == BloomType.ROWPREFIX_FIXED_LENGTH) {
1022      bloomParam = familyDescriptor.getConfigurationValue(BloomFilterUtil.PREFIX_LENGTH_KEY);
1023    }
1024    return bloomParam;
1025  };
1026
1027  /**
1028   * Serialize column family to data block encoding map to configuration. Invoked while configuring
1029   * the MR job for incremental load.
1030   */
1031  @InterfaceAudience.Private
1032  static Function<ColumnFamilyDescriptor, String> dataBlockEncodingDetails = familyDescriptor -> {
1033    DataBlockEncoding encoding = familyDescriptor.getDataBlockEncoding();
1034    if (encoding == null) {
1035      encoding = DataBlockEncoding.NONE;
1036    }
1037    return encoding.toString();
1038  };
1039
1040}