| /** |
| * 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.tajo.storage.rcfile; |
| |
| import org.apache.commons.lang.StringEscapeUtils; |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.*; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.io.*; |
| import org.apache.hadoop.io.SequenceFile.Metadata; |
| import org.apache.hadoop.io.compress.*; |
| import org.apache.hadoop.util.ReflectionUtils; |
| import org.apache.tajo.catalog.Column; |
| import org.apache.tajo.catalog.Schema; |
| import org.apache.tajo.catalog.TableMeta; |
| import org.apache.tajo.catalog.proto.CatalogProtos; |
| import org.apache.tajo.catalog.statistics.TableStats; |
| import org.apache.tajo.conf.TajoConf; |
| import org.apache.tajo.datum.Datum; |
| import org.apache.tajo.datum.NullDatum; |
| import org.apache.tajo.storage.*; |
| import org.apache.tajo.storage.fragment.FileFragment; |
| import org.apache.tajo.tuple.offheap.OffHeapRowBlock; |
| import org.apache.tajo.tuple.offheap.OffHeapRowBlockWriter; |
| import org.apache.tajo.util.ReflectionUtil; |
| |
| import java.io.Closeable; |
| import java.io.*; |
| import java.rmi.server.UID; |
| import java.security.MessageDigest; |
| import java.util.Arrays; |
| |
| /** |
| * <code>RCFile</code>s, short of Record Columnar File, are flat files |
| * consisting of binary key/value pairs, which shares much similarity with |
| * <code>SequenceFile</code>. |
| * <p/> |
| * RCFile stores columns of a table in a record columnar way. It first |
| * partitions rows horizontally into row splits. and then it vertically |
| * partitions each row split in a columnar way. RCFile first stores the meta |
| * data of a row split, as the key part of a record, and all the data of a row |
| * split as the value part. When writing, RCFile.Writer first holds records' |
| * value bytes in memory, and determines a row split if the raw bytes size of |
| * buffered records overflow a given parameter<tt>Writer.columnsBufferSize</tt>, |
| * which can be set like: <code>conf.setInt(COLUMNS_BUFFER_SIZE_CONF_STR, |
| * 4 * 1024 * 1024)</code> . |
| * <p> |
| * <code>RCFile</code> provides {@link Writer}, {@link Reader} and classes for |
| * writing, reading respectively. |
| * </p> |
| * <p/> |
| * <p> |
| * RCFile stores columns of a table in a record columnar way. It first |
| * partitions rows horizontally into row splits. and then it vertically |
| * partitions each row split in a columnar way. RCFile first stores the meta |
| * data of a row split, as the key part of a record, and all the data of a row |
| * split as the value part. |
| * </p> |
| * <p/> |
| * <p> |
| * RCFile compresses values in a more fine-grained manner then record level |
| * compression. However, It currently does not support compress the key part |
| * yet. The actual compression algorithm used to compress key and/or values can |
| * be specified by using the appropriate {@link CompressionCodec}. |
| * </p> |
| * <p/> |
| * <p> |
| * The {@link Reader} is used to read and explain the bytes of RCFile. |
| * </p> |
| * <p/> |
| * <h4 id="Formats">RCFile Formats</h4> |
| * <p/> |
| * <p/> |
| * <h5 id="Header">RC Header</h5> |
| * <ul> |
| * <li>version - 3 bytes of magic header <b>RCF</b>, followed by 1 byte of |
| * actual version number (e.g. RCF1)</li> |
| * <li>compression - A boolean which specifies if compression is turned on for |
| * keys/values in this file.</li> |
| * <li>compression codec - <code>CompressionCodec</code> class which is used |
| * for compression of keys and/or values (if compression is enabled).</li> |
| * <li>metadata - {@link Metadata} for this file.</li> |
| * <li>sync - A sync marker to denote end of the header.</li> |
| * </ul> |
| * <p/> |
| * <h5>RCFile Format</h5> |
| * <ul> |
| * <li><a href="#Header">Header</a></li> |
| * <li>Record |
| * <li>Key part |
| * <ul> |
| * <li>Record length in bytes</li> |
| * <li>Key length in bytes</li> |
| * <li>Number_of_rows_in_this_record(vint)</li> |
| * <li>Column_1_ondisk_length(vint)</li> |
| * <li>Column_1_row_1_value_plain_length</li> |
| * <li>Column_1_row_2_value_plain_length</li> |
| * <li>...</li> |
| * <li>Column_2_ondisk_length(vint)</li> |
| * <li>Column_2_row_1_value_plain_length</li> |
| * <li>Column_2_row_2_value_plain_length</li> |
| * <li>...</li> |
| * </ul> |
| * </li> |
| * </li> |
| * <li>Value part |
| * <ul> |
| * <li>Compressed or plain data of [column_1_row_1_value, |
| * column_1_row_2_value,....]</li> |
| * <li>Compressed or plain data of [column_2_row_1_value, |
| * column_2_row_2_value,....]</li> |
| * </ul> |
| * </li> |
| * </ul> |
| * <p> |
| * <pre> |
| * {@code |
| * The following is a pseudo-BNF grammar for RCFile. Comments are prefixed |
| * with dashes: |
| * |
| * rcfile ::= |
| * <file-header> |
| * <rcfile-rowgroup>+ |
| * |
| * file-header ::= |
| * <file-version-header> |
| * <file-key-class-name> (only exists if version is seq6) |
| * <file-value-class-name> (only exists if version is seq6) |
| * <file-is-compressed> |
| * <file-is-block-compressed> (only exists if version is seq6) |
| * [<file-compression-codec-class>] |
| * <file-header-metadata> |
| * <file-sync-field> |
| * |
| * -- The normative RCFile implementation included with Hive is actually |
| * -- based on a modified version of Hadoop's SequenceFile code. Some |
| * -- things which should have been modified were not, including the code |
| * -- that writes out the file version header. Consequently, RCFile and |
| * -- SequenceFile originally shared the same version header. A newer |
| * -- release has created a unique version string. |
| * |
| * file-version-header ::= Byte[4] {'S', 'E', 'Q', 6} |
| * | Byte[4] {'R', 'C', 'F', 1} |
| * |
| * -- The name of the Java class responsible for reading the key buffer |
| * -- component of the rowgroup. |
| * |
| * file-key-class-name ::= |
| * Text {"org.apache.hadoop.hive.ql.io.RCFile$KeyBuffer"} |
| * |
| * -- The name of the Java class responsible for reading the value buffer |
| * -- component of the rowgroup. |
| * |
| * file-value-class-name ::= |
| * Text {"org.apache.hadoop.hive.ql.io.RCFile$ValueBuffer"} |
| * |
| * -- Boolean variable indicating whether or not the file uses compression |
| * -- for the key and column buffer sections. |
| * |
| * file-is-compressed ::= Byte[1] |
| * |
| * -- A boolean field indicating whether or not the file is block compressed. |
| * -- This field is *always* false. According to comments in the original |
| * -- RCFile implementation this field was retained for backwards |
| * -- compatability with the SequenceFile format. |
| * |
| * file-is-block-compressed ::= Byte[1] {false} |
| * |
| * -- The Java class name of the compression codec iff <file-is-compressed> |
| * -- is true. The named class must implement |
| * -- org.apache.hadoop.io.compress.CompressionCodec. |
| * -- The expected value is org.apache.hadoop.io.compress.GzipCodec. |
| * |
| * file-compression-codec-class ::= Text |
| * |
| * -- A collection of key-value pairs defining metadata values for the |
| * -- file. The Map is serialized using standard JDK serialization, i.e. |
| * -- an Int corresponding to the number of key-value pairs, followed by |
| * -- Text key and value pairs. The following metadata properties are |
| * -- mandatory for all RCFiles: |
| * -- |
| * -- hive.io.rcfile.column.number: the number of columns in the RCFile |
| * |
| * file-header-metadata ::= Map<Text, Text> |
| * |
| * -- A 16 byte marker that is generated by the writer. This marker appears |
| * -- at regular intervals at the beginning of rowgroup-headers, and is |
| * -- intended to enable readers to skip over corrupted rowgroups. |
| * |
| * file-sync-hash ::= Byte[16] |
| * |
| * -- Each row group is split into three sections: a header, a set of |
| * -- key buffers, and a set of column buffers. The header section includes |
| * -- an optional sync hash, information about the size of the row group, and |
| * -- the total number of rows in the row group. Each key buffer |
| * -- consists of run-length encoding data which is used to decode |
| * -- the length and offsets of individual fields in the corresponding column |
| * -- buffer. |
| * |
| * rcfile-rowgroup ::= |
| * <rowgroup-header> |
| * <rowgroup-key-data> |
| * <rowgroup-column-buffers> |
| * |
| * rowgroup-header ::= |
| * [<rowgroup-sync-marker>, <rowgroup-sync-hash>] |
| * <rowgroup-record-length> |
| * <rowgroup-key-length> |
| * <rowgroup-compressed-key-length> |
| * |
| * -- rowgroup-key-data is compressed if the column data is compressed. |
| * rowgroup-key-data ::= |
| * <rowgroup-num-rows> |
| * <rowgroup-key-buffers> |
| * |
| * -- An integer (always -1) signaling the beginning of a sync-hash |
| * -- field. |
| * |
| * rowgroup-sync-marker ::= Int |
| * |
| * -- A 16 byte sync field. This must match the <file-sync-hash> value read |
| * -- in the file header. |
| * |
| * rowgroup-sync-hash ::= Byte[16] |
| * |
| * -- The record-length is the sum of the number of bytes used to store |
| * -- the key and column parts, i.e. it is the total length of the current |
| * -- rowgroup. |
| * |
| * rowgroup-record-length ::= Int |
| * |
| * -- Total length in bytes of the rowgroup's key sections. |
| * |
| * rowgroup-key-length ::= Int |
| * |
| * -- Total compressed length in bytes of the rowgroup's key sections. |
| * |
| * rowgroup-compressed-key-length ::= Int |
| * |
| * -- Number of rows in the current rowgroup. |
| * |
| * rowgroup-num-rows ::= VInt |
| * |
| * -- One or more column key buffers corresponding to each column |
| * -- in the RCFile. |
| * |
| * rowgroup-key-buffers ::= <rowgroup-key-buffer>+ |
| * |
| * -- Data in each column buffer is stored using a run-length |
| * -- encoding scheme that is intended to reduce the cost of |
| * -- repeated column field values. This mechanism is described |
| * -- in more detail in the following entries. |
| * |
| * rowgroup-key-buffer ::= |
| * <column-buffer-length> |
| * <column-buffer-uncompressed-length> |
| * <column-key-buffer-length> |
| * <column-key-buffer> |
| * |
| * -- The serialized length on disk of the corresponding column buffer. |
| * |
| * column-buffer-length ::= VInt |
| * |
| * -- The uncompressed length of the corresponding column buffer. This |
| * -- is equivalent to column-buffer-length if the RCFile is not compressed. |
| * |
| * column-buffer-uncompressed-length ::= VInt |
| * |
| * -- The length in bytes of the current column key buffer |
| * |
| * column-key-buffer-length ::= VInt |
| * |
| * -- The column-key-buffer contains a sequence of serialized VInt values |
| * -- corresponding to the byte lengths of the serialized column fields |
| * -- in the corresponding rowgroup-column-buffer. For example, consider |
| * -- an integer column that contains the consecutive values 1, 2, 3, 44. |
| * -- The RCFile format stores these values as strings in the column buffer, |
| * -- e.g. "12344". The length of each column field is recorded in |
| * -- the column-key-buffer as a sequence of VInts: 1,1,1,2. However, |
| * -- if the same length occurs repeatedly, then we replace repeated |
| * -- run lengths with the complement (i.e. negative) of the number of |
| * -- repetitions, so 1,1,1,2 becomes 1,~2,2. |
| * |
| * column-key-buffer ::= Byte[column-key-buffer-length] |
| * |
| * rowgroup-column-buffers ::= <rowgroup-value-buffer>+ |
| * |
| * -- RCFile stores all column data as strings regardless of the |
| * -- underlying column type. The strings are neither length-prefixed or |
| * -- null-terminated, and decoding them into individual fields requires |
| * -- the use of the run-length information contained in the corresponding |
| * -- column-key-buffer. |
| * |
| * rowgroup-column-buffer ::= Byte[column-buffer-length] |
| * |
| * Byte ::= An eight-bit byte |
| * |
| * VInt ::= Variable length integer. The high-order bit of each byte |
| * indicates whether more bytes remain to be read. The low-order seven |
| * bits are appended as increasingly more significant bits in the |
| * resulting integer value. |
| * |
| * Int ::= A four-byte integer in big-endian format. |
| * |
| * Text ::= VInt, Chars (Length prefixed UTF-8 characters) |
| * } |
| * </pre> |
| * </p> |
| */ |
| public class RCFile { |
| |
| private static final Log LOG = LogFactory.getLog(RCFile.class); |
| |
| public static final String RECORD_INTERVAL_CONF_STR = "hive.io.rcfile.record.interval"; |
| public static final String COLUMN_NUMBER_METADATA_STR = "hive.io.rcfile.column.number"; |
| |
| // All of the versions should be place in this list. |
| private static final int ORIGINAL_VERSION = 0; // version with SEQ |
| private static final int NEW_MAGIC_VERSION = 1; // version with RCF |
| |
| private static final int CURRENT_VERSION = NEW_MAGIC_VERSION; |
| |
| // The first version of RCFile used the sequence file header. |
| private static final byte[] ORIGINAL_MAGIC = new byte[]{ |
| (byte) 'S', (byte) 'E', (byte) 'Q'}; |
| // the version that was included with the original magic, which is mapped |
| // into ORIGINAL_VERSION |
| private static final byte ORIGINAL_MAGIC_VERSION_WITH_METADATA = 6; |
| |
| private static final byte[] ORIGINAL_MAGIC_VERSION = new byte[]{ |
| (byte) 'S', (byte) 'E', (byte) 'Q', ORIGINAL_MAGIC_VERSION_WITH_METADATA |
| }; |
| |
| // The 'magic' bytes at the beginning of the RCFile |
| private static final byte[] MAGIC = new byte[]{ |
| (byte) 'R', (byte) 'C', (byte) 'F'}; |
| |
| private static final int SYNC_ESCAPE = -1; // "length" of sync entries |
| private static final int SYNC_HASH_SIZE = 16; // number of bytes in hash |
| private static final int SYNC_SIZE = 4 + SYNC_HASH_SIZE; // escape + hash |
| |
| /** |
| * The number of bytes between sync points. |
| */ |
| public static final int SYNC_INTERVAL = 100 * SYNC_SIZE; |
| public static final String NULL = "rcfile.null"; |
| public static final String SERDE = "rcfile.serde"; |
| |
| /** |
| * KeyBuffer is the key of each record in RCFile. Its on-disk layout is as |
| * below: |
| * <p/> |
| * <ul> |
| * <li>record length in bytes,it is the sum of bytes used to store the key |
| * part and the value part.</li> |
| * <li>Key length in bytes, it is how many bytes used by the key part.</li> |
| * <li>number_of_rows_in_this_record(vint),</li> |
| * <li>column_1_ondisk_length(vint),</li> |
| * <li>column_1_row_1_value_plain_length,</li> |
| * <li>column_1_row_2_value_plain_length,</li> |
| * <li>....</li> |
| * <li>column_2_ondisk_length(vint),</li> |
| * <li>column_2_row_1_value_plain_length,</li> |
| * <li>column_2_row_2_value_plain_length,</li> |
| * <li>.... .</li> |
| * <li>{the end of the key part}</li> |
| * </ul> |
| */ |
| public static class KeyBuffer { |
| // each column's length in the value |
| private int[] eachColumnValueLen = null; |
| private int[] eachColumnUncompressedValueLen = null; |
| // stores each cell's length of a column in one DataOutputBuffer element |
| private NonSyncByteArrayOutputStream[] allCellValLenBuffer = null; |
| // how many rows in this split |
| private int numberRows = 0; |
| // how many columns |
| private int columnNumber = 0; |
| |
| KeyBuffer(int columnNum) { |
| columnNumber = columnNum; |
| eachColumnValueLen = new int[columnNumber]; |
| eachColumnUncompressedValueLen = new int[columnNumber]; |
| allCellValLenBuffer = new NonSyncByteArrayOutputStream[columnNumber]; |
| } |
| |
| public void readFields(DataInput in) throws IOException { |
| eachColumnValueLen = new int[columnNumber]; |
| eachColumnUncompressedValueLen = new int[columnNumber]; |
| allCellValLenBuffer = new NonSyncByteArrayOutputStream[columnNumber]; |
| |
| numberRows = WritableUtils.readVInt(in); |
| for (int i = 0; i < columnNumber; i++) { |
| eachColumnValueLen[i] = WritableUtils.readVInt(in); |
| eachColumnUncompressedValueLen[i] = WritableUtils.readVInt(in); |
| int bufLen = WritableUtils.readVInt(in); |
| if (allCellValLenBuffer[i] == null) { |
| allCellValLenBuffer[i] = new NonSyncByteArrayOutputStream(); |
| } else { |
| allCellValLenBuffer[i].reset(); |
| } |
| allCellValLenBuffer[i].write(in, bufLen); |
| } |
| } |
| |
| /** |
| * @return the numberRows |
| */ |
| public int getNumberRows() { |
| return numberRows; |
| } |
| } |
| |
| /** |
| * ValueBuffer is the value of each record in RCFile. Its on-disk layout is as |
| * below: |
| * <ul> |
| * <li>Compressed or plain data of [column_1_row_1_value, |
| * column_1_row_2_value,....]</li> |
| * <li>Compressed or plain data of [column_2_row_1_value, |
| * column_2_row_2_value,....]</li> |
| * </ul> |
| */ |
| public static class ValueBuffer implements Closeable{ |
| |
| // used to load columns' value into memory |
| private NonSyncByteArrayOutputStream[] loadedColumnsValueBuffer = null; |
| |
| boolean inited = false; |
| |
| // used for readFields |
| KeyBuffer keyBuffer; |
| private int columnNumber = 0; |
| |
| // set true for columns that needed to skip loading into memory. |
| boolean[] skippedColIDs = null; |
| |
| CompressionCodec codec; |
| Decompressor decompressor = null; |
| NonSyncDataInputBuffer decompressBuffer = new NonSyncDataInputBuffer(); |
| private long readBytes = 0; |
| |
| |
| public ValueBuffer(KeyBuffer currentKey, int columnNumber, |
| int[] targets, CompressionCodec codec, boolean[] skippedIDs) |
| throws IOException { |
| keyBuffer = currentKey; |
| this.columnNumber = columnNumber; |
| this.skippedColIDs = skippedIDs; |
| this.codec = codec; |
| loadedColumnsValueBuffer = new NonSyncByteArrayOutputStream[targets.length]; |
| if (codec != null) { |
| decompressor = org.apache.tajo.storage.compress.CodecPool.getDecompressor(codec); |
| } |
| |
| for (int i = 0; i < targets.length; i++) { |
| loadedColumnsValueBuffer[i] = new NonSyncByteArrayOutputStream(); |
| } |
| } |
| |
| public void readFields(DataInput in) throws IOException { |
| int addIndex = 0; |
| int skipTotal = 0; |
| |
| |
| for (int i = 0; i < columnNumber; i++) { |
| int vaRowsLen = keyBuffer.eachColumnValueLen[i]; |
| // skip this column |
| if (skippedColIDs[i]) { |
| skipTotal += vaRowsLen; |
| continue; |
| } |
| |
| if (skipTotal != 0) { |
| StorageUtil.skipFully(in, skipTotal); |
| skipTotal = 0; |
| } |
| |
| NonSyncByteArrayOutputStream valBuf; |
| if (codec != null) { |
| // load into compressed buf first |
| |
| byte[] compressedBytes = new byte[vaRowsLen]; |
| in.readFully(compressedBytes, 0, vaRowsLen); |
| |
| decompressBuffer.reset(compressedBytes, vaRowsLen); |
| if(decompressor != null) decompressor.reset(); |
| |
| DataInputStream is; |
| if (codec instanceof SplittableCompressionCodec) { |
| SplitCompressionInputStream deflatFilter = ((SplittableCompressionCodec) codec).createInputStream( |
| decompressBuffer, decompressor, 0, vaRowsLen, SplittableCompressionCodec.READ_MODE.BYBLOCK); |
| is = new DataInputStream(deflatFilter); |
| } else { |
| CompressionInputStream deflatFilter = codec.createInputStream(decompressBuffer, decompressor); |
| is = new DataInputStream(deflatFilter); |
| } |
| |
| valBuf = loadedColumnsValueBuffer[addIndex]; |
| valBuf.reset(); |
| valBuf.write(is, keyBuffer.eachColumnUncompressedValueLen[i]); |
| is.close(); |
| decompressBuffer.close(); |
| } else { |
| valBuf = loadedColumnsValueBuffer[addIndex]; |
| valBuf.reset(); |
| valBuf.write(in, vaRowsLen); |
| } |
| readBytes += keyBuffer.eachColumnUncompressedValueLen[i]; |
| addIndex++; |
| } |
| |
| if (skipTotal != 0) { |
| StorageUtil.skipFully(in, skipTotal); |
| } |
| } |
| |
| public long getReadBytes() { |
| return readBytes; |
| } |
| |
| public void clearColumnBuffer() throws IOException { |
| decompressBuffer.reset(); |
| readBytes = 0; |
| } |
| |
| @Override |
| public void close() { |
| for (NonSyncByteArrayOutputStream element : loadedColumnsValueBuffer) { |
| IOUtils.closeStream(element); |
| } |
| if (codec != null) { |
| IOUtils.closeStream(decompressBuffer); |
| if (decompressor != null) { |
| // Make sure we only return decompressor once. |
| org.apache.tajo.storage.compress.CodecPool.returnDecompressor(decompressor); |
| decompressor = null; |
| } |
| } |
| } |
| } |
| |
| /** |
| * Create a metadata object with alternating key-value pairs. |
| * Eg. metadata(key1, value1, key2, value2) |
| */ |
| public static Metadata createMetadata(Text... values) { |
| if (values.length % 2 != 0) { |
| throw new IllegalArgumentException("Must have a matched set of " + |
| "key-value pairs. " + values.length + |
| " strings supplied."); |
| } |
| Metadata result = new Metadata(); |
| for (int i = 0; i < values.length; i += 2) { |
| result.set(values[i], values[i + 1]); |
| } |
| return result; |
| } |
| |
| /** |
| * Write KeyBuffer/ValueBuffer pairs to a RCFile. RCFile's format is |
| * compatible with SequenceFile's. |
| */ |
| public static class RCFileAppender extends FileAppender { |
| FSDataOutputStream out; |
| |
| CompressionCodec codec = null; |
| Metadata metadata = null; |
| FileSystem fs = null; |
| TableStatistics stats = null; |
| int columnNumber = 0; |
| |
| // how many records the writer buffers before it writes to disk |
| private int RECORD_INTERVAL = Integer.MAX_VALUE; |
| // the max size of memory for buffering records before writes them out |
| private int COLUMNS_BUFFER_SIZE = 16 * 1024 * 1024; // 16M |
| // the conf string for COLUMNS_BUFFER_SIZE |
| public static final String COLUMNS_BUFFER_SIZE_CONF_STR = "hive.io.rcfile.record.buffer.size"; |
| |
| // how many records already buffered |
| private int bufferedRecords = 0; |
| private ColumnBuffer[] columnBuffers = null; |
| boolean useNewMagic = true; |
| private byte[] nullChars; |
| private SerializerDeserializer serde; |
| private boolean isShuffle; |
| |
| // Insert a globally unique 16-byte value every few entries, so that one |
| // can seek into the middle of a file and then synchronize with record |
| // starts and ends by scanning for this value. |
| long lastSyncPos; // position of last sync |
| byte[] sync; // 16 random bytes |
| |
| { |
| try { |
| MessageDigest digester = MessageDigest.getInstance("MD5"); |
| long time = System.currentTimeMillis(); |
| digester.update((new UID() + "@" + time).getBytes()); |
| sync = digester.digest(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /* |
| * used for buffering appends before flush them out |
| */ |
| class ColumnBuffer { |
| // used for buffer a column's values |
| NonSyncByteArrayOutputStream columnValBuffer; |
| // used to store each value's length |
| NonSyncByteArrayOutputStream valLenBuffer; |
| |
| /* |
| * use a run-length encoding. We only record run length if a same |
| * 'prevValueLen' occurs more than one time. And we negative the run |
| * length to distinguish a runLength and a normal value length. For |
| * example, if the values' lengths are 1,1,1,2, we record 1, ~2,2. And for |
| * value lengths 1,2,3 we record 1,2,3. |
| */ |
| int columnValueLength = 0; |
| int uncompressedColumnValueLength = 0; |
| int columnKeyLength = 0; |
| int runLength = 0; |
| int prevValueLength = -1; |
| |
| ColumnBuffer() throws IOException { |
| columnValBuffer = new NonSyncByteArrayOutputStream(); |
| valLenBuffer = new NonSyncByteArrayOutputStream(); |
| } |
| |
| public int append(Column column, Datum datum) throws IOException { |
| int currentLen = serde.serialize(column, datum, columnValBuffer, nullChars); |
| columnValueLength += currentLen; |
| uncompressedColumnValueLength += currentLen; |
| |
| if (prevValueLength < 0) { |
| startNewGroup(currentLen); |
| return currentLen; |
| } |
| |
| if (currentLen != prevValueLength) { |
| flushGroup(); |
| startNewGroup(currentLen); |
| } else { |
| runLength++; |
| } |
| return currentLen; |
| } |
| |
| private void startNewGroup(int currentLen) { |
| prevValueLength = currentLen; |
| runLength = 0; |
| } |
| |
| public void clear() { |
| valLenBuffer.reset(); |
| columnValBuffer.reset(); |
| prevValueLength = -1; |
| runLength = 0; |
| columnValueLength = 0; |
| columnKeyLength = 0; |
| uncompressedColumnValueLength = 0; |
| } |
| |
| public int flushGroup() { |
| int len = 0; |
| if (prevValueLength >= 0) { |
| len += valLenBuffer.writeVLong(prevValueLength); |
| if (runLength > 0) { |
| len += valLenBuffer.writeVLong(~runLength); |
| } |
| columnKeyLength += len; |
| runLength = -1; |
| prevValueLength = -1; |
| } |
| return len; |
| } |
| |
| public int UnFlushedGroupSize() { |
| int len = 0; |
| if (prevValueLength >= 0) { |
| len += WritableUtils.getVIntSize(prevValueLength); |
| if (runLength > 0) { |
| len += WritableUtils.getVIntSize(~runLength); |
| } |
| } |
| return len; |
| } |
| } |
| |
| public long getLength() throws IOException { |
| return out.getPos(); |
| } |
| |
| public RCFileAppender(Configuration conf, final Schema schema, final TableMeta meta, final Path path) throws IOException { |
| super(conf, schema, meta, path); |
| |
| RECORD_INTERVAL = conf.getInt(RECORD_INTERVAL_CONF_STR, RECORD_INTERVAL); |
| COLUMNS_BUFFER_SIZE = conf.getInt(COLUMNS_BUFFER_SIZE_CONF_STR, COLUMNS_BUFFER_SIZE); |
| columnNumber = schema.size(); |
| } |
| |
| public void init() throws IOException { |
| fs = path.getFileSystem(conf); |
| |
| if (!fs.exists(path.getParent())) { |
| throw new FileNotFoundException(path.toString()); |
| } |
| |
| //determine the intermediate file type |
| String store = conf.get(TajoConf.ConfVars.SHUFFLE_FILE_FORMAT.varname, |
| TajoConf.ConfVars.SHUFFLE_FILE_FORMAT.defaultVal); |
| if (enabledStats && CatalogProtos.StoreType.RCFILE == CatalogProtos.StoreType.valueOf(store.toUpperCase())) { |
| isShuffle = true; |
| } else { |
| isShuffle = false; |
| } |
| |
| if (this.meta.containsOption(StorageConstants.COMPRESSION_CODEC)) { |
| String codecClassname = this.meta.getOption(StorageConstants.COMPRESSION_CODEC); |
| try { |
| Class<? extends CompressionCodec> codecClass = conf.getClassByName( |
| codecClassname).asSubclass(CompressionCodec.class); |
| codec = ReflectionUtil.newInstance(codecClass); |
| } catch (ClassNotFoundException cnfe) { |
| throw new IllegalArgumentException( |
| "Unknown codec: " + codecClassname, cnfe); |
| } |
| } |
| |
| String nullCharacters = StringEscapeUtils.unescapeJava(this.meta.getOption(StorageConstants.RCFILE_NULL, |
| NullDatum.DEFAULT_TEXT)); |
| if (StringUtils.isEmpty(nullCharacters)) { |
| nullChars = NullDatum.get().asTextBytes(); |
| } else { |
| nullChars = nullCharacters.getBytes(); |
| } |
| |
| if (metadata == null) { |
| metadata = new Metadata(); |
| } |
| |
| metadata.set(new Text(COLUMN_NUMBER_METADATA_STR), new Text("" + columnNumber)); |
| |
| String serdeClass = this.meta.getOption(StorageConstants.RCFILE_SERDE, |
| BinarySerializerDeserializer.class.getName()); |
| try { |
| serde = (SerializerDeserializer) ReflectionUtil.newInstance(Class.forName(serdeClass)); |
| } catch (Exception e) { |
| LOG.error(e.getMessage(), e); |
| throw new IOException(e); |
| } |
| metadata.set(new Text(StorageConstants.RCFILE_SERDE), new Text(serdeClass)); |
| |
| columnBuffers = new ColumnBuffer[columnNumber]; |
| for (int i = 0; i < columnNumber; i++) { |
| columnBuffers[i] = new ColumnBuffer(); |
| } |
| |
| init(conf, fs.create(path, true, 4096, (short) 3, fs.getDefaultBlockSize(), null), codec, metadata); |
| initializeFileHeader(); |
| writeFileHeader(); |
| finalizeFileHeader(); |
| |
| if (enabledStats) { |
| this.stats = new TableStatistics(this.schema); |
| } |
| super.init(); |
| } |
| |
| /** |
| * Write the initial part of file header. |
| */ |
| void initializeFileHeader() throws IOException { |
| if (useNewMagic) { |
| out.write(MAGIC); |
| out.write(CURRENT_VERSION); |
| } else { |
| out.write(ORIGINAL_MAGIC_VERSION); |
| } |
| } |
| |
| /** |
| * Write the final part of file header. |
| */ |
| void finalizeFileHeader() throws IOException { |
| out.write(sync); // write the sync bytes |
| out.flush(); // flush header |
| } |
| |
| boolean isCompressed() { |
| return codec != null; |
| } |
| |
| /** |
| * Write and flush the file header. |
| */ |
| void writeFileHeader() throws IOException { |
| if (useNewMagic) { |
| out.writeBoolean(isCompressed()); |
| } else { |
| Text.writeString(out, "org.apache.hadoop.hive.ql.io.RCFile$KeyBuffer"); |
| Text.writeString(out, "org.apache.hadoop.hive.ql.io.RCFile$ValueBuffer"); |
| out.writeBoolean(isCompressed()); |
| out.writeBoolean(false); |
| } |
| |
| if (isCompressed()) { |
| Text.writeString(out, (codec.getClass()).getName()); |
| } |
| metadata.write(out); |
| } |
| |
| void init(Configuration conf, FSDataOutputStream out, |
| CompressionCodec codec, Metadata metadata) throws IOException { |
| this.out = out; |
| this.codec = codec; |
| this.metadata = metadata; |
| this.useNewMagic = conf.getBoolean(TajoConf.ConfVars.HIVEUSEEXPLICITRCFILEHEADER.varname, true); |
| } |
| |
| /** |
| * create a sync point. |
| */ |
| public void sync() throws IOException { |
| if (sync != null && lastSyncPos != out.getPos()) { |
| out.writeInt(SYNC_ESCAPE); // mark the start of the sync |
| out.write(sync); // write sync |
| lastSyncPos = out.getPos(); // update lastSyncPos |
| } |
| } |
| |
| private void checkAndWriteSync() throws IOException { |
| if (sync != null && out.getPos() >= lastSyncPos + SYNC_INTERVAL) { |
| sync(); |
| } |
| } |
| |
| private int columnBufferSize = 0; |
| |
| @Override |
| public long getOffset() throws IOException { |
| return out.getPos(); |
| } |
| |
| @Override |
| public void flush() throws IOException { |
| flushRecords(); |
| out.flush(); |
| } |
| |
| @Override |
| public void addTuple(Tuple t) throws IOException { |
| append(t); |
| // Statistical section |
| |
| if (enabledStats) { |
| stats.incrementRow(); |
| } |
| } |
| |
| /** |
| * Append a row of values. Currently it only can accept < |
| * {@link Tuple}. If its <code>size()</code> is less than the |
| * column number in the file, zero bytes are appended for the empty columns. |
| * If its size() is greater then the column number in the file, the exceeded |
| * columns' bytes are ignored. |
| * |
| * @param tuple a Tuple with the list of serialized columns |
| * @throws IOException |
| */ |
| public void append(Tuple tuple) throws IOException { |
| int size = schema.size(); |
| |
| for (int i = 0; i < size; i++) { |
| Datum datum = tuple.get(i); |
| int length = columnBuffers[i].append(schema.getColumn(i), datum); |
| columnBufferSize += length; |
| if (isShuffle) { |
| // it is to calculate min/max values, and it is only used for the intermediate file. |
| stats.analyzeField(i, datum); |
| } |
| } |
| |
| if (size < columnNumber) { |
| for (int i = size; i < columnNumber; i++) { |
| columnBuffers[i].append(schema.getColumn(i), NullDatum.get()); |
| if (isShuffle) { |
| stats.analyzeField(i, NullDatum.get()); |
| } |
| } |
| } |
| |
| bufferedRecords++; |
| //TODO compression rate base flush |
| if ((columnBufferSize > COLUMNS_BUFFER_SIZE) |
| || (bufferedRecords >= RECORD_INTERVAL)) { |
| flushRecords(); |
| } |
| } |
| |
| /** |
| * get number of bytes to store the keyBuffer. |
| * |
| * @return number of bytes used to store this KeyBuffer on disk |
| * @throws IOException |
| */ |
| public int getKeyBufferSize() throws IOException { |
| int ret = 0; |
| ret += WritableUtils.getVIntSize(bufferedRecords); |
| for (int i = 0; i < columnBuffers.length; i++) { |
| ColumnBuffer currentBuf = columnBuffers[i]; |
| ret += WritableUtils.getVIntSize(currentBuf.columnValueLength); |
| ret += WritableUtils.getVIntSize(currentBuf.uncompressedColumnValueLength); |
| ret += WritableUtils.getVIntSize(currentBuf.columnKeyLength); |
| ret += currentBuf.columnKeyLength; |
| } |
| |
| return ret; |
| } |
| |
| /** |
| * get number of bytes to store the key part. |
| * |
| * @return number of bytes used to store this Key part on disk |
| * @throws IOException |
| */ |
| public int getKeyPartSize() throws IOException { |
| int ret = 12; //12 bytes |record count, key length, compressed key length| |
| |
| ret += WritableUtils.getVIntSize(bufferedRecords); |
| for (int i = 0; i < columnBuffers.length; i++) { |
| ColumnBuffer currentBuf = columnBuffers[i]; |
| ret += WritableUtils.getVIntSize(currentBuf.columnValueLength); |
| ret += WritableUtils.getVIntSize(currentBuf.uncompressedColumnValueLength); |
| ret += WritableUtils.getVIntSize(currentBuf.columnKeyLength); |
| ret += currentBuf.columnKeyLength; |
| ret += currentBuf.UnFlushedGroupSize(); |
| } |
| |
| return ret; |
| } |
| |
| private void WriteKeyBuffer(DataOutputStream out) throws IOException { |
| WritableUtils.writeVLong(out, bufferedRecords); |
| for (int i = 0; i < columnBuffers.length; i++) { |
| ColumnBuffer currentBuf = columnBuffers[i]; |
| WritableUtils.writeVLong(out, currentBuf.columnValueLength); |
| WritableUtils.writeVLong(out, currentBuf.uncompressedColumnValueLength); |
| WritableUtils.writeVLong(out, currentBuf.columnKeyLength); |
| currentBuf.valLenBuffer.writeTo(out); |
| } |
| } |
| |
| private void flushRecords() throws IOException { |
| |
| Compressor compressor = null; |
| NonSyncByteArrayOutputStream valueBuffer = null; |
| CompressionOutputStream deflateFilter = null; |
| DataOutputStream deflateOut = null; |
| boolean isCompressed = isCompressed(); |
| |
| int valueLength = 0; |
| if (isCompressed) { |
| compressor = org.apache.tajo.storage.compress.CodecPool.getCompressor(codec); |
| if (compressor != null) compressor.reset(); //builtin gzip is null |
| |
| valueBuffer = new NonSyncByteArrayOutputStream(); |
| deflateFilter = codec.createOutputStream(valueBuffer, compressor); |
| deflateOut = new DataOutputStream(deflateFilter); |
| } |
| |
| try { |
| for (int columnIndex = 0; columnIndex < columnNumber; columnIndex++) { |
| ColumnBuffer currentBuf = columnBuffers[columnIndex]; |
| currentBuf.flushGroup(); |
| |
| NonSyncByteArrayOutputStream columnValue = currentBuf.columnValBuffer; |
| int colLen; |
| int plainLen = columnValue.getLength(); |
| if (isCompressed) { |
| deflateFilter.resetState(); |
| deflateOut.write(columnValue.getData(), 0, columnValue.getLength()); |
| deflateOut.flush(); |
| deflateFilter.finish(); |
| columnValue.close(); |
| // find how much compressed data was added for this column |
| colLen = valueBuffer.getLength() - valueLength; |
| currentBuf.columnValueLength = colLen; |
| } else { |
| colLen = plainLen; |
| } |
| valueLength += colLen; |
| } |
| } catch (IOException e) { |
| IOUtils.cleanup(LOG, deflateOut, out); |
| throw e; |
| } |
| |
| if (compressor != null) { |
| org.apache.tajo.storage.compress.CodecPool.returnCompressor(compressor); |
| } |
| |
| int keyLength = getKeyBufferSize(); |
| if (keyLength < 0) { |
| throw new IOException("negative length keys not allowed: " + keyLength); |
| } |
| // Write the key out |
| writeKey(keyLength + valueLength, keyLength); |
| // write the value out |
| if (isCompressed) { |
| try { |
| out.write(valueBuffer.getData(), 0, valueBuffer.getLength()); |
| } finally { |
| IOUtils.cleanup(LOG, valueBuffer); |
| } |
| } else { |
| for (int columnIndex = 0; columnIndex < columnNumber; ++columnIndex) { |
| columnBuffers[columnIndex].columnValBuffer.writeTo(out); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Column#" + columnIndex + " : Plain Total Column Value Length: " |
| + columnBuffers[columnIndex].uncompressedColumnValueLength |
| + ", Compr Total Column Value Length: " + columnBuffers[columnIndex].columnValueLength); |
| } |
| } |
| } |
| // clear the columnBuffers |
| clearColumnBuffers(); |
| |
| bufferedRecords = 0; |
| columnBufferSize = 0; |
| } |
| |
| private void writeKey(int recordLen, int keyLength) throws IOException { |
| checkAndWriteSync(); // sync |
| out.writeInt(recordLen); // total record length |
| out.writeInt(keyLength); // key portion length |
| |
| if (this.isCompressed()) { |
| Compressor compressor = org.apache.tajo.storage.compress.CodecPool.getCompressor(codec); |
| if (compressor != null) compressor.reset(); //builtin gzip is null |
| |
| NonSyncByteArrayOutputStream compressionBuffer = new NonSyncByteArrayOutputStream(); |
| CompressionOutputStream deflateFilter = codec.createOutputStream(compressionBuffer, compressor); |
| DataOutputStream deflateOut = new DataOutputStream(deflateFilter); |
| |
| //compress key and write key out |
| compressionBuffer.reset(); |
| deflateFilter.resetState(); |
| WriteKeyBuffer(deflateOut); |
| deflateOut.flush(); |
| deflateFilter.finish(); |
| int compressedKeyLen = compressionBuffer.getLength(); |
| out.writeInt(compressedKeyLen); |
| compressionBuffer.writeTo(out); |
| compressionBuffer.reset(); |
| deflateOut.close(); |
| org.apache.tajo.storage.compress.CodecPool.returnCompressor(compressor); |
| } else { |
| out.writeInt(keyLength); |
| WriteKeyBuffer(out); |
| } |
| } |
| |
| private void clearColumnBuffers() throws IOException { |
| for (int i = 0; i < columnNumber; i++) { |
| columnBuffers[i].clear(); |
| } |
| } |
| |
| @Override |
| public TableStats getStats() { |
| if (enabledStats) { |
| return stats.getTableStat(); |
| } else { |
| return null; |
| } |
| } |
| |
| @Override |
| public void close() throws IOException { |
| if (bufferedRecords > 0) { |
| flushRecords(); |
| } |
| clearColumnBuffers(); |
| |
| if (out != null) { |
| // Statistical section |
| if (enabledStats) { |
| stats.setNumBytes(getOffset()); |
| } |
| // Close the underlying stream if we own it... |
| out.flush(); |
| IOUtils.cleanup(LOG, out); |
| out = null; |
| } |
| } |
| } |
| |
| /** |
| * Read KeyBuffer/ValueBuffer pairs from a RCFile. |
| */ |
| public static class RCFileScanner extends FileScanner { |
| private static class SelectedColumn { |
| public int colIndex; |
| public int rowReadIndex; |
| public int runLength; |
| public int prvLength; |
| public boolean isNulled; |
| } |
| |
| private FSDataInputStream in; |
| |
| private byte version; |
| |
| private CompressionCodec codec = null; |
| private Metadata metadata = null; |
| |
| private byte[] sync; |
| private byte[] syncCheck; |
| private boolean syncSeen; |
| private long lastSeenSyncPos = 0; |
| |
| private long headerEnd; |
| private long start, end; |
| private final long startOffset, endOffset; |
| private int[] targetColumnIndexes; |
| |
| private int currentKeyLength; |
| private int currentRecordLength; |
| |
| private ValueBuffer currentValue; |
| |
| private int readRowsIndexInBuffer = 0; |
| |
| private int recordsNumInValBuffer = 0; |
| |
| private int columnNumber = 0; |
| |
| private boolean more = true; |
| |
| private int passedRowsNum = 0; |
| |
| private boolean decompress = false; |
| |
| private Decompressor keyDecompressor; |
| |
| private long readBytes = 0; |
| |
| //Current state of each selected column - e.g. current run length, etc. |
| // The size of the array is equal to the number of selected columns |
| private SelectedColumn[] selectedColumns; |
| |
| // column value lengths for each of the selected columns |
| private NonSyncDataInputBuffer[] colValLenBufferReadIn; |
| |
| private LongWritable rowId; |
| private byte[] nullChars; |
| private SerializerDeserializer serde; |
| |
| public RCFileScanner(Configuration conf, final Schema schema, final TableMeta meta, |
| final FileFragment fragment) throws IOException { |
| super(conf, schema, meta, fragment); |
| conf.setInt("io.file.buffer.size", 4096); //TODO remove |
| |
| startOffset = fragment.getStartKey(); |
| endOffset = startOffset + fragment.getEndKey(); |
| start = 0; |
| } |
| |
| @Override |
| public void init() throws IOException { |
| sync = new byte[SYNC_HASH_SIZE]; |
| syncCheck = new byte[SYNC_HASH_SIZE]; |
| |
| more = startOffset < endOffset; |
| rowId = new LongWritable(); |
| readBytes = 0; |
| |
| String nullCharacters = StringEscapeUtils.unescapeJava(meta.getOption(StorageConstants.RCFILE_NULL, |
| NullDatum.DEFAULT_TEXT)); |
| if (StringUtils.isEmpty(nullCharacters)) { |
| nullChars = NullDatum.get().asTextBytes(); |
| } else { |
| nullChars = nullCharacters.getBytes(); |
| } |
| |
| // projection |
| if (targets == null) { |
| targets = schema.toArray(); |
| } |
| |
| targetColumnIndexes = new int[targets.length]; |
| for (int i = 0; i < targets.length; i++) { |
| targetColumnIndexes[i] = schema.getColumnId(targets[i].getQualifiedName()); |
| } |
| Arrays.sort(targetColumnIndexes); |
| |
| FileSystem fs = fragment.getPath().getFileSystem(conf); |
| end = fs.getFileStatus(fragment.getPath()).getLen(); |
| in = openFile(fs, fragment.getPath(), 4096); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("RCFile open:" + fragment.getPath() + "," + start + "," + (endOffset - startOffset) + |
| "," + fs.getFileStatus(fragment.getPath()).getLen()); |
| } |
| //init RCFILE Header |
| boolean succeed = false; |
| try { |
| if (start > 0) { |
| seek(0); |
| initHeader(); |
| } else { |
| initHeader(); |
| } |
| succeed = true; |
| } finally { |
| if (!succeed) { |
| if (in != null) { |
| try { |
| in.close(); |
| } catch (IOException e) { |
| if (LOG != null && LOG.isDebugEnabled()) { |
| LOG.debug("Exception in closing " + in, e); |
| } |
| } |
| } |
| } |
| } |
| |
| columnNumber = Integer.parseInt(metadata.get(new Text(COLUMN_NUMBER_METADATA_STR)).toString()); |
| selectedColumns = new SelectedColumn[targetColumnIndexes.length]; |
| colValLenBufferReadIn = new NonSyncDataInputBuffer[targetColumnIndexes.length]; |
| boolean[] skippedColIDs = new boolean[columnNumber]; |
| Arrays.fill(skippedColIDs, true); |
| super.init(); |
| |
| for (int i = 0; i < targetColumnIndexes.length; i++) { |
| int tid = targetColumnIndexes[i]; |
| if (tid < columnNumber) { |
| skippedColIDs[tid] = false; |
| |
| SelectedColumn col = new SelectedColumn(); |
| col.colIndex = tid; |
| col.runLength = 0; |
| col.prvLength = -1; |
| col.rowReadIndex = 0; |
| selectedColumns[i] = col; |
| colValLenBufferReadIn[i] = new NonSyncDataInputBuffer(); |
| } |
| } |
| |
| currentKey = createKeyBuffer(); |
| currentValue = new ValueBuffer(null, columnNumber, targetColumnIndexes, codec, skippedColIDs); |
| |
| if (startOffset > getPosition()) { // TODO use sync cache |
| sync(startOffset); // sync to start |
| } |
| } |
| |
| /** |
| * Return the metadata (Text to Text map) that was written into the |
| * file. |
| */ |
| public Metadata getMetadata() { |
| return metadata; |
| } |
| |
| /** |
| * Return the metadata value associated with the given key. |
| * |
| * @param key the metadata key to retrieve |
| */ |
| public Text getMetadataValueOf(Text key) { |
| return metadata.get(key); |
| } |
| |
| /** |
| * Override this method to specialize the type of |
| * {@link FSDataInputStream} returned. |
| */ |
| protected FSDataInputStream openFile(FileSystem fs, Path file, int bufferSize) throws IOException { |
| return fs.open(file, bufferSize); |
| } |
| |
| private void initHeader() throws IOException { |
| byte[] magic = new byte[MAGIC.length]; |
| in.readFully(magic); |
| |
| if (Arrays.equals(magic, ORIGINAL_MAGIC)) { |
| byte vers = in.readByte(); |
| if (vers != ORIGINAL_MAGIC_VERSION_WITH_METADATA) { |
| throw new IOException(fragment.getPath() + " is a version " + vers + |
| " SequenceFile instead of an RCFile."); |
| } |
| version = ORIGINAL_VERSION; |
| } else { |
| if (!Arrays.equals(magic, MAGIC)) { |
| throw new IOException(fragment.getPath() + " not a RCFile and has magic of " + |
| new String(magic)); |
| } |
| |
| // Set 'version' |
| version = in.readByte(); |
| if (version > CURRENT_VERSION) { |
| throw new VersionMismatchException((byte) CURRENT_VERSION, version); |
| } |
| } |
| |
| if (version == ORIGINAL_VERSION) { |
| try { |
| Class<?> keyCls = conf.getClassByName(Text.readString(in)); |
| Class<?> valCls = conf.getClassByName(Text.readString(in)); |
| if (!keyCls.equals(KeyBuffer.class) |
| || !valCls.equals(ValueBuffer.class)) { |
| throw new IOException(fragment.getPath() + " not a RCFile"); |
| } |
| } catch (ClassNotFoundException e) { |
| throw new IOException(fragment.getPath() + " not a RCFile", e); |
| } |
| } |
| |
| decompress = in.readBoolean(); // is compressed? |
| |
| if (version == ORIGINAL_VERSION) { |
| // is block-compressed? it should be always false. |
| boolean blkCompressed = in.readBoolean(); |
| if (blkCompressed) { |
| throw new IOException(fragment.getPath() + " not a RCFile."); |
| } |
| } |
| |
| // setup the compression codec |
| if (decompress) { |
| String codecClassname = Text.readString(in); |
| try { |
| Class<? extends CompressionCodec> codecClass = conf.getClassByName( |
| codecClassname).asSubclass(CompressionCodec.class); |
| codec = ReflectionUtils.newInstance(codecClass, conf); |
| } catch (ClassNotFoundException cnfe) { |
| throw new IllegalArgumentException( |
| "Unknown codec: " + codecClassname, cnfe); |
| } |
| |
| keyDecompressor = org.apache.tajo.storage.compress.CodecPool.getDecompressor(codec); |
| } |
| |
| metadata = new Metadata(); |
| metadata.readFields(in); |
| |
| Text text = metadata.get(new Text(StorageConstants.RCFILE_SERDE)); |
| |
| try { |
| String serdeClass; |
| if(text != null && !text.toString().isEmpty()){ |
| serdeClass = text.toString(); |
| } else{ |
| serdeClass = this.meta.getOption(StorageConstants.RCFILE_SERDE, BinarySerializerDeserializer.class.getName()); |
| } |
| serde = (SerializerDeserializer) ReflectionUtil.newInstance(Class.forName(serdeClass)); |
| } catch (Exception e) { |
| LOG.error(e.getMessage(), e); |
| throw new IOException(e); |
| } |
| |
| in.readFully(sync); // read sync bytes |
| headerEnd = in.getPos(); |
| lastSeenSyncPos = headerEnd; //initial sync position |
| readBytes += headerEnd; |
| } |
| |
| /** |
| * Return the current byte position in the input file. |
| */ |
| public long getPosition() throws IOException { |
| return in.getPos(); |
| } |
| |
| /** |
| * Set the current byte position in the input file. |
| * <p/> |
| * <p/> |
| * The position passed must be a position returned by |
| * {@link RCFile.RCFileAppender#getLength()} when writing this file. To seek to an |
| * arbitrary position, use {@link RCFile.RCFileScanner#sync(long)}. In another |
| * words, the current seek can only seek to the end of the file. For other |
| * positions, use {@link RCFile.RCFileScanner#sync(long)}. |
| */ |
| public void seek(long position) throws IOException { |
| in.seek(position); |
| } |
| |
| /** |
| * Resets the values which determine if there are more rows in the buffer |
| * <p/> |
| * This can be used after one calls seek or sync, if one called next before that. |
| * Otherwise, the seek or sync will have no effect, it will continue to get rows from the |
| * buffer built up from the call to next. |
| */ |
| public void resetBuffer() { |
| readRowsIndexInBuffer = 0; |
| recordsNumInValBuffer = 0; |
| } |
| |
| /** |
| * Seek to the next sync mark past a given position. |
| */ |
| public void sync(long position) throws IOException { |
| if (position + SYNC_SIZE >= end) { |
| seek(end); |
| return; |
| } |
| |
| //this is to handle syn(pos) where pos < headerEnd. |
| if (position < headerEnd) { |
| // seek directly to first record |
| in.seek(headerEnd); |
| // note the sync marker "seen" in the header |
| syncSeen = true; |
| return; |
| } |
| |
| try { |
| seek(position + 4); // skip escape |
| |
| int prefix = sync.length; |
| int n = conf.getInt("io.bytes.per.checksum", 512); |
| byte[] buffer = new byte[prefix + n]; |
| n = (int) Math.min(n, end - in.getPos()); |
| /* fill array with a pattern that will never match sync */ |
| Arrays.fill(buffer, (byte) (~sync[0])); |
| while (n > 0 && (in.getPos() + n) <= end) { |
| position = in.getPos(); |
| in.readFully(buffer, prefix, n); |
| readBytes += n; |
| /* the buffer has n+sync bytes */ |
| for (int i = 0; i < n; i++) { |
| int j; |
| for (j = 0; j < sync.length && sync[j] == buffer[i + j]; j++) { |
| /* nothing */ |
| } |
| if (j == sync.length) { |
| /* simplified from (position + (i - prefix) + sync.length) - SYNC_SIZE */ |
| in.seek(position + i - SYNC_SIZE); |
| return; |
| } |
| } |
| /* move the last 16 bytes to the prefix area */ |
| System.arraycopy(buffer, buffer.length - prefix, buffer, 0, prefix); |
| n = (int) Math.min(n, end - in.getPos()); |
| } |
| } catch (ChecksumException e) { // checksum failure |
| handleChecksumException(e); |
| } |
| } |
| |
| private void handleChecksumException(ChecksumException e) throws IOException { |
| if (conf.getBoolean("io.skip.checksum.errors", false)) { |
| LOG.warn("Bad checksum at " + getPosition() + ". Skipping entries."); |
| sync(getPosition() + conf.getInt("io.bytes.per.checksum", 512)); |
| } else { |
| throw e; |
| } |
| } |
| |
| private KeyBuffer createKeyBuffer() { |
| return new KeyBuffer(columnNumber); |
| } |
| |
| /** |
| * Read and return the next record length, potentially skipping over a sync |
| * block. |
| * |
| * @return the length of the next record or -1 if there is no next record |
| * @throws IOException |
| */ |
| private int readRecordLength() throws IOException { |
| if (in.getPos() >= end) { |
| return -1; |
| } |
| int length = in.readInt(); |
| readBytes += 4; |
| if (sync != null && length == SYNC_ESCAPE) { // process |
| // a |
| // sync entry |
| lastSeenSyncPos = in.getPos() - 4; // minus SYNC_ESCAPE's length |
| in.readFully(syncCheck); // read syncCheck |
| readBytes += SYNC_HASH_SIZE; |
| if (!Arrays.equals(sync, syncCheck)) { |
| throw new IOException("File is corrupt!"); |
| } |
| syncSeen = true; |
| if (in.getPos() >= end) { |
| return -1; |
| } |
| length = in.readInt(); // re-read length |
| readBytes += 4; |
| } else { |
| syncSeen = false; |
| } |
| return length; |
| } |
| |
| private void seekToNextKeyBuffer() throws IOException { |
| if (!keyInit) { |
| return; |
| } |
| if (!currentValue.inited) { |
| IOUtils.skipFully(in, currentRecordLength - currentKeyLength); |
| } |
| } |
| |
| private int compressedKeyLen = 0; |
| NonSyncDataInputBuffer keyDataIn = new NonSyncDataInputBuffer(); |
| NonSyncDataInputBuffer keyDecompressBuffer = new NonSyncDataInputBuffer(); |
| |
| KeyBuffer currentKey = null; |
| boolean keyInit = false; |
| |
| protected int nextKeyBuffer() throws IOException { |
| seekToNextKeyBuffer(); |
| currentRecordLength = readRecordLength(); |
| if (currentRecordLength == -1) { |
| keyInit = false; |
| return -1; |
| } |
| currentKeyLength = in.readInt(); |
| compressedKeyLen = in.readInt(); |
| readBytes += 8; |
| if (decompress) { |
| |
| byte[] compressedBytes = new byte[compressedKeyLen]; |
| in.readFully(compressedBytes, 0, compressedKeyLen); |
| |
| if (keyDecompressor != null) keyDecompressor.reset(); |
| keyDecompressBuffer.reset(compressedBytes, compressedKeyLen); |
| |
| DataInputStream is; |
| if (codec instanceof SplittableCompressionCodec) { |
| SplitCompressionInputStream deflatFilter = ((SplittableCompressionCodec) codec).createInputStream( |
| keyDecompressBuffer, keyDecompressor, 0, compressedKeyLen, SplittableCompressionCodec.READ_MODE.BYBLOCK); |
| |
| keyDecompressBuffer.seek(deflatFilter.getAdjustedStart()); |
| is = new DataInputStream(deflatFilter); |
| } else { |
| CompressionInputStream deflatFilter = codec.createInputStream(keyDecompressBuffer, keyDecompressor); |
| is = new DataInputStream(deflatFilter); |
| } |
| |
| byte[] deCompressedBytes = new byte[currentKeyLength]; |
| |
| is.readFully(deCompressedBytes, 0, currentKeyLength); |
| keyDataIn.reset(deCompressedBytes, currentKeyLength); |
| currentKey.readFields(keyDataIn); |
| is.close(); |
| } else { |
| currentKey.readFields(in); |
| } |
| readBytes += currentKeyLength; |
| keyInit = true; |
| currentValue.inited = false; |
| |
| readRowsIndexInBuffer = 0; |
| recordsNumInValBuffer = currentKey.numberRows; |
| |
| for (int selIx = 0; selIx < selectedColumns.length; selIx++) { |
| SelectedColumn col = selectedColumns[selIx]; |
| if (col == null) { |
| col = new SelectedColumn(); |
| col.isNulled = true; |
| selectedColumns[selIx] = col; |
| continue; |
| } |
| |
| int colIx = col.colIndex; |
| NonSyncByteArrayOutputStream buf = currentKey.allCellValLenBuffer[colIx]; |
| colValLenBufferReadIn[selIx].reset(buf.getData(), buf.getLength()); |
| col.rowReadIndex = 0; |
| col.runLength = 0; |
| col.prvLength = -1; |
| col.isNulled = buf.getLength() == 0; |
| } |
| |
| return currentKeyLength; |
| } |
| |
| protected void currentValueBuffer() throws IOException { |
| if (!keyInit) { |
| nextKeyBuffer(); |
| } |
| currentValue.keyBuffer = currentKey; |
| currentValue.clearColumnBuffer(); |
| currentValue.readFields(in); |
| currentValue.inited = true; |
| readBytes += currentValue.getReadBytes(); |
| |
| if (tableStats != null) { |
| tableStats.setReadBytes(readBytes); |
| tableStats.setNumRows(passedRowsNum); |
| } |
| } |
| |
| private boolean rowFetched = false; |
| |
| @Override |
| public Tuple next() throws IOException { |
| if(!hasNext()) return null; |
| |
| Tuple tuple = new VTuple(schema.size()); |
| getCurrentRow(tuple); |
| return tuple; |
| } |
| |
| public boolean hasNext() throws IOException { |
| if (!more) { |
| return false; |
| } |
| |
| more = nextBuffer(rowId); |
| long lastSeenSyncPos = lastSeenSyncPos(); |
| if (lastSeenSyncPos >= endOffset) { |
| more = false; |
| return more; |
| } |
| |
| return more; |
| } |
| |
| @Override |
| public boolean nextFetch(OffHeapRowBlock rowBlock) throws IOException { |
| rowBlock.clear(); |
| |
| OffHeapRowBlockWriter writer = (OffHeapRowBlockWriter) rowBlock.getWriter(); |
| while(rowBlock.rows() < rowBlock.maxRowNum() && hasNext()) { |
| getCurrentRowBlock(writer); |
| } |
| |
| return rowBlock.rows() > 0; |
| } |
| |
| @Override |
| public float getProgress() { |
| try { |
| if(!more) { |
| return 1.0f; |
| } |
| long filePos = getPosition(); |
| if (startOffset == filePos) { |
| return 0.0f; |
| } else { |
| //if scanner read the header, filePos moved to zero |
| return Math.min(1.0f, (float)(Math.max(filePos - startOffset, 0)) / (float)(fragment.getEndKey())); |
| } |
| } catch (IOException e) { |
| LOG.error(e.getMessage(), e); |
| return 0.0f; |
| } |
| } |
| |
| /** |
| * Returns how many rows we fetched with nextBuffer(). It only means how many rows |
| * are read by nextBuffer(). The returned result may be smaller than actual number |
| * of rows passed by, because {@link #seek(long)} can change the underlying key buffer and |
| * value buffer. |
| * |
| * @return next row number |
| * @throws IOException |
| */ |
| public boolean nextBuffer(LongWritable readRows) throws IOException { |
| if (readRowsIndexInBuffer < recordsNumInValBuffer) { |
| readRows.set(passedRowsNum); |
| readRowsIndexInBuffer++; |
| passedRowsNum++; |
| rowFetched = false; |
| return true; |
| } else { |
| keyInit = false; |
| } |
| |
| int ret = -1; |
| try { |
| ret = nextKeyBuffer(); |
| } catch (EOFException eof) { |
| eof.printStackTrace(); |
| } |
| return (ret > 0) && nextBuffer(readRows); |
| } |
| |
| /** |
| * get the current row used,make sure called {@link #next()} |
| * first. |
| * |
| * @throws IOException |
| */ |
| public void getCurrentRow(Tuple tuple) throws IOException { |
| if (!keyInit || rowFetched) { |
| return; |
| } |
| |
| if (!currentValue.inited) { |
| currentValueBuffer(); |
| } |
| |
| for (int j = 0; j < selectedColumns.length; ++j) { |
| SelectedColumn col = selectedColumns[j]; |
| int i = col.colIndex; |
| |
| if (col.isNulled) { |
| tuple.put(i, NullDatum.get()); |
| } else { |
| colAdvanceRow(j, col); |
| |
| Datum datum = serde.deserialize(schema.getColumn(i), |
| currentValue.loadedColumnsValueBuffer[j].getData(), col.rowReadIndex, col.prvLength, nullChars); |
| tuple.put(i, datum); |
| col.rowReadIndex += col.prvLength; |
| } |
| } |
| rowFetched = true; |
| } |
| |
| /** |
| * get the current row used,make sure called {@link #nextFetch(OffHeapRowBlock rowBlock)} |
| * first. |
| * |
| * @throws IOException |
| */ |
| public void getCurrentRowBlock(OffHeapRowBlockWriter writer) throws IOException { |
| if (!keyInit || rowFetched) { |
| return; |
| } |
| |
| if (!currentValue.inited) { |
| currentValueBuffer(); |
| } |
| writer.startRow(); |
| int currentSchemaIndex = 0; |
| for (int j = 0; j < selectedColumns.length; ++j) { |
| SelectedColumn col = selectedColumns[j]; |
| int i = col.colIndex; |
| |
| while(i > currentSchemaIndex){ |
| writer.skipField(); |
| currentSchemaIndex++; |
| } |
| |
| if (col.isNulled) { |
| writer.skipField(); |
| } else { |
| colAdvanceRow(j, col); |
| |
| serde.write(writer, schema.getColumn(i), |
| currentValue.loadedColumnsValueBuffer[j].getData(), col.rowReadIndex, col.prvLength, nullChars); |
| col.rowReadIndex += col.prvLength; |
| } |
| currentSchemaIndex++; |
| } |
| |
| while(schema.size() > currentSchemaIndex){ |
| writer.skipField(); |
| currentSchemaIndex++; |
| } |
| writer.endRow(); |
| rowFetched = true; |
| } |
| |
| /** |
| * Advance column state to the next now: update offsets, run lengths etc |
| * |
| * @param selCol - index among selectedColumns |
| * @param col - column object to update the state of. prvLength will be |
| * set to the new read position |
| * @throws IOException |
| */ |
| private void colAdvanceRow(int selCol, SelectedColumn col) throws IOException { |
| if (col.runLength > 0) { |
| --col.runLength; |
| } else { |
| int length = (int) WritableUtils.readVLong(colValLenBufferReadIn[selCol]); |
| if (length < 0) { |
| // we reach a runlength here, use the previous length and reset |
| // runlength |
| col.runLength = (~length) - 1; |
| } else { |
| col.prvLength = length; |
| col.runLength = 0; |
| } |
| } |
| } |
| |
| /** |
| * Returns true if the previous call to next passed a sync mark. |
| */ |
| public boolean syncSeen() { |
| return syncSeen; |
| } |
| |
| /** |
| * Returns the last seen sync position. |
| */ |
| public long lastSeenSyncPos() { |
| return lastSeenSyncPos; |
| } |
| |
| /** |
| * Returns the name of the file. |
| */ |
| @Override |
| public String toString() { |
| return fragment.getPath().toString(); |
| } |
| |
| @Override |
| public void reset() throws IOException { |
| seek(startOffset); |
| } |
| |
| @Override |
| public boolean isProjectable() { |
| return true; |
| } |
| |
| @Override |
| public boolean isSelectable() { |
| return false; |
| } |
| |
| @Override |
| public boolean isSplittable() { |
| return true; |
| } |
| |
| @Override |
| public void close() throws IOException { |
| if (tableStats != null) { |
| tableStats.setReadBytes(readBytes); //Actual Processed Bytes. (decompressed bytes + header - seek) |
| tableStats.setNumRows(passedRowsNum); |
| } |
| |
| IOUtils.cleanup(LOG, in, currentValue); |
| if (keyDecompressor != null) { |
| // Make sure we only return decompressor once. |
| org.apache.tajo.storage.compress.CodecPool.returnDecompressor(keyDecompressor); |
| keyDecompressor = null; |
| } |
| } |
| } |
| } |
