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.codec;
019
020import java.io.IOException;
021import java.io.InputStream;
022import java.io.OutputStream;
023
024import org.apache.commons.io.IOUtils;
025import org.apache.hadoop.hbase.Cell;
026import org.apache.hadoop.hbase.CellBuilderType;
027import org.apache.hadoop.hbase.ExtendedCellBuilder;
028import org.apache.hadoop.hbase.ExtendedCellBuilderFactory;
029import org.apache.hadoop.hbase.HBaseInterfaceAudience;
030import org.apache.hadoop.hbase.io.ByteBuffInputStream;
031import org.apache.hadoop.hbase.nio.ByteBuff;
032import org.apache.hadoop.hbase.util.Bytes;
033import org.apache.yetus.audience.InterfaceAudience;
034
035/**
036 * Basic Cell codec that just writes out all the individual elements of a Cell.  Uses ints
037 * delimiting all lengths. Profligate. Needs tune up.
038 * Note: This will not write tags of a Cell.
039 */
040@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.CONFIG)
041public class CellCodec implements Codec {
042  static class CellEncoder extends BaseEncoder {
043    CellEncoder(final OutputStream out) {
044      super(out);
045    }
046
047    @Override
048    public void write(Cell cell) throws IOException {
049      checkFlushed();
050      // Row
051      write(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
052      // Column family
053      write(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
054      // Qualifier
055      write(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
056      // Version
057      this.out.write(Bytes.toBytes(cell.getTimestamp()));
058      // Type
059      this.out.write(cell.getTypeByte());
060      // Value
061      write(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
062      // MvccVersion
063      this.out.write(Bytes.toBytes(cell.getSequenceId()));
064    }
065
066    /**
067     * Write int length followed by array bytes.
068     * @param bytes
069     * @param offset
070     * @param length
071     * @throws IOException
072     */
073    private void write(final byte [] bytes, final int offset, final int length)
074    throws IOException {
075      // TODO add BB backed os check and do for write. Pass Cell
076      this.out.write(Bytes.toBytes(length));
077      this.out.write(bytes, offset, length);
078    }
079  }
080
081  static class CellDecoder extends BaseDecoder {
082    private final ExtendedCellBuilder cellBuilder = ExtendedCellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
083    public CellDecoder(final InputStream in) {
084      super(in);
085    }
086
087    @Override
088    protected Cell parseCell() throws IOException {
089      byte [] row = readByteArray(this.in);
090      byte [] family = readByteArray(in);
091      byte [] qualifier = readByteArray(in);
092      byte [] longArray = new byte[Bytes.SIZEOF_LONG];
093      IOUtils.readFully(this.in, longArray);
094      long timestamp = Bytes.toLong(longArray);
095      byte type = (byte) this.in.read();
096      byte[] value = readByteArray(in);
097      // Read memstore version
098      byte[] memstoreTSArray = new byte[Bytes.SIZEOF_LONG];
099      IOUtils.readFully(this.in, memstoreTSArray);
100      long memstoreTS = Bytes.toLong(memstoreTSArray);
101      return cellBuilder.clear()
102              .setRow(row)
103              .setFamily(family)
104              .setQualifier(qualifier)
105              .setTimestamp(timestamp)
106              .setType(type)
107              .setValue(value)
108              .setSequenceId(memstoreTS)
109              .build();
110    }
111
112    /**
113     * @return Byte array read from the stream.
114     * @throws IOException
115     */
116    private byte [] readByteArray(final InputStream in) throws IOException {
117      byte [] intArray = new byte[Bytes.SIZEOF_INT];
118      IOUtils.readFully(in, intArray);
119      int length = Bytes.toInt(intArray);
120      byte [] bytes = new byte [length];
121      IOUtils.readFully(in, bytes);
122      return bytes;
123    }
124  }
125
126  @Override
127  public Decoder getDecoder(InputStream is) {
128    return new CellDecoder(is);
129  }
130
131  @Override
132  public Decoder getDecoder(ByteBuff buf) {
133    return getDecoder(new ByteBuffInputStream(buf));
134  }
135
136  @Override
137  public Encoder getEncoder(OutputStream os) {
138    return new CellEncoder(os);
139  }
140}