/** * 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. */ #include "Adaptor.hh" #include "Compression.hh" #include "RLEv2.hh" #include "RLEV2Util.hh" namespace orc { int64_t RleDecoderV2::readLongBE(uint64_t bsz) { int64_t ret = 0, val; uint64_t n = bsz; while (n > 0) { n--; val = readByte(); ret |= (val << (n * 8)); } return ret; } inline int64_t RleDecoderV2::readVslong() { return unZigZag(readVulong()); } uint64_t RleDecoderV2::readVulong() { uint64_t ret = 0, b; uint64_t offset = 0; do { b = readByte(); ret |= (0x7f & b) << offset; offset += 7; } while (b >= 0x80); return ret; } RleDecoderV2::RleDecoderV2(std::unique_ptr input, bool _isSigned, MemoryPool& pool ): inputStream(std::move(input)), isSigned(_isSigned), firstByte(0), runLength(0), runRead(0), bufferStart(nullptr), bufferEnd(bufferStart), deltaBase(0), byteSize(0), firstValue(0), prevValue(0), bitSize(0), bitsLeft(0), curByte(0), patchBitSize(0), unpackedIdx(0), patchIdx(0), base(0), curGap(0), curPatch(0), patchMask(0), actualGap(0), unpacked(pool, 0), unpackedPatch(pool, 0) { // PASS } void RleDecoderV2::seek(PositionProvider& location) { // move the input stream inputStream->seek(location); // clear state bufferEnd = bufferStart = nullptr; runRead = runLength = 0; // skip ahead the given number of records skip(location.next()); } void RleDecoderV2::skip(uint64_t numValues) { // simple for now, until perf tests indicate something encoding specific is // needed const uint64_t N = 64; int64_t dummy[N]; while (numValues) { uint64_t nRead = std::min(N, numValues); next(dummy, nRead, nullptr); numValues -= nRead; } } void RleDecoderV2::next(int64_t* const data, const uint64_t numValues, const char* const notNull) { uint64_t nRead = 0; while (nRead < numValues) { // Skip any nulls before attempting to read first byte. while (notNull && !notNull[nRead]) { if (++nRead == numValues) { return; // ended with null values } } if (runRead == runLength) { resetRun(); firstByte = readByte(); } uint64_t offset = nRead, length = numValues - nRead; EncodingType enc = static_cast ((firstByte >> 6) & 0x03); switch(static_cast(enc)) { case SHORT_REPEAT: nRead += nextShortRepeats(data, offset, length, notNull); break; case DIRECT: nRead += nextDirect(data, offset, length, notNull); break; case PATCHED_BASE: nRead += nextPatched(data, offset, length, notNull); break; case DELTA: nRead += nextDelta(data, offset, length, notNull); break; default: throw ParseError("unknown encoding"); } } } uint64_t RleDecoderV2::nextShortRepeats(int64_t* const data, uint64_t offset, uint64_t numValues, const char* const notNull) { if (runRead == runLength) { // extract the number of fixed bytes byteSize = (firstByte >> 3) & 0x07; byteSize += 1; runLength = firstByte & 0x07; // run lengths values are stored only after MIN_REPEAT value is met runLength += MIN_REPEAT; runRead = 0; // read the repeated value which is store using fixed bytes firstValue = readLongBE(byteSize); if (isSigned) { firstValue = unZigZag(static_cast(firstValue)); } } uint64_t nRead = std::min(runLength - runRead, numValues); if (notNull) { for(uint64_t pos = offset; pos < offset + nRead; ++pos) { if (notNull[pos]) { data[pos] = firstValue; ++runRead; } } } else { for(uint64_t pos = offset; pos < offset + nRead; ++pos) { data[pos] = firstValue; ++runRead; } } return nRead; } uint64_t RleDecoderV2::nextDirect(int64_t* const data, uint64_t offset, uint64_t numValues, const char* const notNull) { if (runRead == runLength) { // extract the number of fixed bits unsigned char fbo = (firstByte >> 1) & 0x1f; bitSize = decodeBitWidth(fbo); // extract the run length runLength = static_cast(firstByte & 0x01) << 8; runLength |= readByte(); // runs are one off runLength += 1; runRead = 0; } uint64_t nRead = std::min(runLength - runRead, numValues); runRead += readLongs(data, offset, nRead, bitSize, notNull); if (isSigned) { if (notNull) { for (uint64_t pos = offset; pos < offset + nRead; ++pos) { if (notNull[pos]) { data[pos] = unZigZag(static_cast(data[pos])); } } } else { for (uint64_t pos = offset; pos < offset + nRead; ++pos) { data[pos] = unZigZag(static_cast(data[pos])); } } } return nRead; } uint64_t RleDecoderV2::nextPatched(int64_t* const data, uint64_t offset, uint64_t numValues, const char* const notNull) { if (runRead == runLength) { // extract the number of fixed bits unsigned char fbo = (firstByte >> 1) & 0x1f; bitSize = decodeBitWidth(fbo); // extract the run length runLength = static_cast(firstByte & 0x01) << 8; runLength |= readByte(); // runs are one off runLength += 1; runRead = 0; // extract the number of bytes occupied by base uint64_t thirdByte = readByte(); byteSize = (thirdByte >> 5) & 0x07; // base width is one off byteSize += 1; // extract patch width uint32_t pwo = thirdByte & 0x1f; patchBitSize = decodeBitWidth(pwo); // read fourth byte and extract patch gap width uint64_t fourthByte = readByte(); uint32_t pgw = (fourthByte >> 5) & 0x07; // patch gap width is one off pgw += 1; // extract the length of the patch list size_t pl = fourthByte & 0x1f; if (pl == 0) { throw ParseError("Corrupt PATCHED_BASE encoded data (pl==0)!"); } // read the next base width number of bytes to extract base value base = readLongBE(byteSize); int64_t mask = (static_cast(1) << ((byteSize * 8) - 1)); // if mask of base value is 1 then base is negative value else positive if ((base & mask) != 0) { base = base & ~mask; base = -base; } // TODO: something more efficient than resize unpacked.resize(runLength); unpackedIdx = 0; readLongs(unpacked.data(), 0, runLength, bitSize); // any remaining bits are thrown out resetReadLongs(); // TODO: something more efficient than resize unpackedPatch.resize(pl); patchIdx = 0; // TODO: Skip corrupt? // if ((patchBitSize + pgw) > 64 && !skipCorrupt) { if ((patchBitSize + pgw) > 64) { throw ParseError("Corrupt PATCHED_BASE encoded data " "(patchBitSize + pgw > 64)!"); } uint32_t cfb = getClosestFixedBits(patchBitSize + pgw); readLongs(unpackedPatch.data(), 0, pl, cfb); // any remaining bits are thrown out resetReadLongs(); // apply the patch directly when decoding the packed data patchMask = ((static_cast(1) << patchBitSize) - 1); adjustGapAndPatch(); } uint64_t nRead = std::min(runLength - runRead, numValues); for(uint64_t pos = offset; pos < offset + nRead; ++pos) { // skip null positions if (notNull && !notNull[pos]) { continue; } if (static_cast(unpackedIdx) != actualGap) { // no patching required. add base to unpacked value to get final value data[pos] = base + unpacked[unpackedIdx]; } else { // extract the patch value int64_t patchedVal = unpacked[unpackedIdx] | (curPatch << bitSize); // add base to patched value data[pos] = base + patchedVal; // increment the patch to point to next entry in patch list ++patchIdx; if (patchIdx < unpackedPatch.size()) { adjustGapAndPatch(); // next gap is relative to the current gap actualGap += unpackedIdx; } } ++runRead; ++unpackedIdx; } return nRead; } uint64_t RleDecoderV2::nextDelta(int64_t* const data, uint64_t offset, uint64_t numValues, const char* const notNull) { if (runRead == runLength) { // extract the number of fixed bits unsigned char fbo = (firstByte >> 1) & 0x1f; if (fbo != 0) { bitSize = decodeBitWidth(fbo); } else { bitSize = 0; } // extract the run length runLength = static_cast(firstByte & 0x01) << 8; runLength |= readByte(); ++runLength; // account for first value runRead = deltaBase = 0; // read the first value stored as vint if (isSigned) { firstValue = static_cast(readVslong()); } else { firstValue = static_cast(readVulong()); } prevValue = firstValue; // read the fixed delta value stored as vint (deltas can be negative even // if all number are positive) deltaBase = static_cast(readVslong()); } uint64_t nRead = std::min(runLength - runRead, numValues); uint64_t pos = offset; for ( ; pos < offset + nRead; ++pos) { // skip null positions if (!notNull || notNull[pos]) break; } if (runRead == 0 && pos < offset + nRead) { data[pos++] = firstValue; ++runRead; } if (bitSize == 0) { // add fixed deltas to adjacent values for ( ; pos < offset + nRead; ++pos) { // skip null positions if (notNull && !notNull[pos]) { continue; } prevValue = data[pos] = prevValue + deltaBase; ++runRead; } } else { for ( ; pos < offset + nRead; ++pos) { // skip null positions if (!notNull || notNull[pos]) break; } if (runRead < 2 && pos < offset + nRead) { // add delta base and first value prevValue = data[pos++] = firstValue + deltaBase; ++runRead; } // write the unpacked values, add it to previous value and store final // value to result buffer. if the delta base value is negative then it // is a decreasing sequence else an increasing sequence uint64_t remaining = (offset + nRead) - pos; runRead += readLongs(data, pos, remaining, bitSize, notNull); if (deltaBase < 0) { for ( ; pos < offset + nRead; ++pos) { // skip null positions if (notNull && !notNull[pos]) { continue; } prevValue = data[pos] = prevValue - data[pos]; } } else { for ( ; pos < offset + nRead; ++pos) { // skip null positions if (notNull && !notNull[pos]) { continue; } prevValue = data[pos] = prevValue + data[pos]; } } } return nRead; } } // namespace orc