001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
003 * agreements. See the NOTICE file distributed with this work for additional information regarding
004 * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the
005 * "License"); you may not use this file except in compliance with the License. You may obtain a
006 * copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable
007 * law or agreed to in writing, software distributed under the License is distributed on an "AS IS"
008 * BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License
009 * for the specific language governing permissions and limitations under the License.
010 */
011package org.apache.hadoop.hbase.io.encoding;
012
013import java.io.DataOutputStream;
014import java.io.IOException;
015
016import org.apache.hadoop.hbase.Cell;
017import org.apache.hadoop.hbase.CellComparatorImpl;
018import org.apache.hadoop.hbase.KeyValueUtil;
019import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
020import org.apache.yetus.audience.InterfaceAudience;
021import org.slf4j.Logger;
022import org.slf4j.LoggerFactory;
023
024@InterfaceAudience.Private
025public class RowIndexEncoderV1 {
026  private static final Logger LOG = LoggerFactory.getLogger(RowIndexEncoderV1.class);
027
028  /** The Cell previously appended. */
029  private Cell lastCell = null;
030
031  private DataOutputStream out;
032  private NoneEncoder encoder;
033  private int startOffset = -1;
034  private ByteArrayOutputStream rowsOffsetBAOS = new ByteArrayOutputStream(64 * 4);
035
036  public RowIndexEncoderV1(DataOutputStream out, HFileBlockDefaultEncodingContext encodingCtx) {
037    this.out = out;
038    this.encoder = new NoneEncoder(out, encodingCtx);
039  }
040
041  public int write(Cell cell) throws IOException {
042    // checkRow uses comparator to check we are writing in order.
043    if (!checkRow(cell)) {
044      if (startOffset < 0) {
045        startOffset = out.size();
046      }
047      rowsOffsetBAOS.writeInt(out.size() - startOffset);
048    }
049    lastCell = cell;
050    return encoder.write(cell);
051  }
052
053  protected boolean checkRow(final Cell cell) throws IOException {
054    boolean isDuplicateRow = false;
055    if (cell == null) {
056      throw new IOException("Key cannot be null or empty");
057    }
058    if (lastCell != null) {
059      int keyComp = CellComparatorImpl.COMPARATOR.compareRows(lastCell, cell);
060      if (keyComp > 0) {
061        throw new IOException("Added a key not lexically larger than"
062            + " previous. Current cell = " + cell + ", lastCell = " + lastCell);
063      } else if (keyComp == 0) {
064        isDuplicateRow = true;
065      }
066    }
067    return isDuplicateRow;
068  }
069
070  public void flush() throws IOException {
071    int onDiskDataSize = 0;
072    if (startOffset >= 0) {
073      onDiskDataSize = out.size() - startOffset;
074    }
075    out.writeInt(rowsOffsetBAOS.size() / 4);
076    if (rowsOffsetBAOS.size() > 0) {
077      out.write(rowsOffsetBAOS.getBuffer(), 0, rowsOffsetBAOS.size());
078    }
079    out.writeInt(onDiskDataSize);
080    if (LOG.isTraceEnabled()) {
081      LOG.trace("RowNumber: " + rowsOffsetBAOS.size() / 4
082          + ", onDiskDataSize: " + onDiskDataSize + ", totalOnDiskSize: "
083          + (out.size() - startOffset));
084    }
085  }
086
087  void beforeShipped() {
088    if (this.lastCell != null) {
089      this.lastCell = KeyValueUtil.toNewKeyCell(this.lastCell);
090    }
091  }
092}