c++/src/RLEv1.cc - orc - Git at Google

 /**
  * 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 "orc/Exceptions.hh"
 #include "RLEv1.hh"

 #include <algorithm>

 namespace orc {

 const uint64_t MINIMUM_REPEAT = 3;
 const uint64_t MAXIMUM_REPEAT = 127 + MINIMUM_REPEAT;

 const int64_t BASE_128_MASK = 0x7f;

 const int64_t MAX_DELTA = 127;
 const int64_t MIN_DELTA = -128;
 const uint64_t MAX_LITERAL_SIZE = 128;

 RleEncoderV1::RleEncoderV1(
                           std::unique_ptr<BufferedOutputStream> outStream,
                           bool hasSigned):
                           RleEncoder(std::move(outStream), hasSigned) {
   literals = new int64_t[MAX_LITERAL_SIZE];
   delta = 0;
   repeat = false;
   tailRunLength = 0;
 }

 RleEncoderV1::~RleEncoderV1() {
   delete [] literals;
 }

 void RleEncoderV1::writeValues() {
   if (numLiterals != 0) {
     if (repeat) {
       writeByte(static_cast<char>
                 (static_cast<uint64_t>(numLiterals) - MINIMUM_REPEAT));
       writeByte(static_cast<char>(delta));
       if (isSigned) {
         writeVslong(literals[0]);
       } else {
         writeVulong(literals[0]);
       }
     } else {
       writeByte(static_cast<char>(-numLiterals));
       for(size_t i=0; i < numLiterals; ++i) {
         if (isSigned) {
           writeVslong(literals[i]);
         } else {
           writeVulong(literals[i]);
         }
       }
     }
     repeat = false;
     numLiterals = 0;
     tailRunLength = 0;
   }
 }

 uint64_t RleEncoderV1::flush() {
   writeValues();
   outputStream->BackUp(static_cast<int>(bufferLength - bufferPosition));
   uint64_t dataSize = outputStream->flush();
   bufferLength = bufferPosition = 0;
   return dataSize;
 }

 void RleEncoderV1::write(int64_t value) {
   if (numLiterals == 0) {
     literals[numLiterals++] = value;
     tailRunLength = 1;
   } else if (repeat) {
     if (value == literals[0] + delta * static_cast<int64_t>(numLiterals)) {
       numLiterals += 1;
       if (numLiterals == MAXIMUM_REPEAT) {
         writeValues();
       }
     } else {
       writeValues();
       literals[numLiterals++] = value;
       tailRunLength = 1;
     }
   } else {
     if (tailRunLength == 1) {
       delta = value - literals[numLiterals - 1];
       if (delta < MIN_DELTA || delta > MAX_DELTA) {
         tailRunLength = 1;
       } else {
         tailRunLength = 2;
       }
     } else if (value == literals[numLiterals - 1] + delta) {
       tailRunLength += 1;
     } else {
       delta = value - literals[numLiterals - 1];
       if (delta < MIN_DELTA || delta > MAX_DELTA) {
         tailRunLength = 1;
       } else {
         tailRunLength = 2;
       }
     }
     if (tailRunLength == MINIMUM_REPEAT) {
       if (numLiterals + 1 == MINIMUM_REPEAT) {
         repeat = true;
         numLiterals += 1;
       } else {
         numLiterals -= static_cast<int>(MINIMUM_REPEAT - 1);
         int64_t base = literals[numLiterals];
         writeValues();
         literals[0] = base;
         repeat = true;
         numLiterals = MINIMUM_REPEAT;
       }
     } else {
       literals[numLiterals++] = value;
       if (numLiterals == MAX_LITERAL_SIZE) {
         writeValues();
       }
     }
   }
 }

 signed char RleDecoderV1::readByte() {
   if (bufferStart == bufferEnd) {
     int bufferLength;
     const void* bufferPointer;
     if (!inputStream->Next(&bufferPointer, &bufferLength)) {
       throw ParseError("bad read in readByte");
     }
     bufferStart = static_cast<const char*>(bufferPointer);
     bufferEnd = bufferStart + bufferLength;
   }
   return *(bufferStart++);
 }

 uint64_t RleDecoderV1::readLong() {
   uint64_t result = 0;
   int64_t offset = 0;
   signed char ch = readByte();
   if (ch >= 0) {
     result = static_cast<uint64_t>(ch);
   } else {
     result = static_cast<uint64_t>(ch) & BASE_128_MASK;
     while ((ch = readByte()) < 0) {
       offset += 7;
       result |= (static_cast<uint64_t>(ch) & BASE_128_MASK) << offset;
     }
     result |= static_cast<uint64_t>(ch) << (offset + 7);
   }
   return result;
 }

 void RleDecoderV1::skipLongs(uint64_t numValues) {
   while (numValues > 0) {
     if (readByte() >= 0) {
       --numValues;
     }
   }
 }

 void RleDecoderV1::readHeader() {
   signed char ch = readByte();
   if (ch < 0) {
     remainingValues = static_cast<uint64_t>(-ch);
     repeating = false;
   } else {
     remainingValues = static_cast<uint64_t>(ch) + MINIMUM_REPEAT;
     repeating = true;
     delta = readByte();
     value = isSigned
         ? unZigZag(readLong())
         : static_cast<int64_t>(readLong());
   }
 }

 RleDecoderV1::RleDecoderV1(std::unique_ptr<SeekableInputStream> input,
                            bool hasSigned)
     : inputStream(std::move(input)),
       isSigned(hasSigned),
       remainingValues(0),
       value(0),
       bufferStart(nullptr),
       bufferEnd(bufferStart),
       delta(0),
       repeating(false) {
 }

 void RleDecoderV1::seek(PositionProvider& location) {
   // move the input stream
   inputStream->seek(location);
   // force a re-read from the stream
   bufferEnd = bufferStart;
   // read a new header
   readHeader();
   // skip ahead the given number of records
   skip(location.next());
 }

 void RleDecoderV1::skip(uint64_t numValues) {
   while (numValues > 0) {
     if (remainingValues == 0) {
       readHeader();
     }
     uint64_t count = std::min(numValues, remainingValues);
     remainingValues -= count;
     numValues -= count;
     if (repeating) {
       value += delta * static_cast<int64_t>(count);
     } else {
       skipLongs(count);
     }
   }
 }

 void RleDecoderV1::next(int64_t* const data,
                         const uint64_t numValues,
                         const char* const notNull) {
   uint64_t position = 0;
   // skipNulls()
   if (notNull) {
     // Skip over null values.
     while (position < numValues && !notNull[position]) {
       ++position;
     }
   }
   while (position < numValues) {
     // If we are out of values, read more.
     if (remainingValues == 0) {
       readHeader();
     }
     // How many do we read out of this block?
     uint64_t count = std::min(numValues - position, remainingValues);
     uint64_t consumed = 0;
     if (repeating) {
       if (notNull) {
         for (uint64_t i = 0; i < count; ++i) {
           if (notNull[position + i]) {
             data[position + i] = value + static_cast<int64_t>(consumed) * delta;
             consumed += 1;
           }
         }
       } else {
         for (uint64_t i = 0; i < count; ++i) {
           data[position + i] = value + static_cast<int64_t>(i) * delta;
         }
         consumed = count;
       }
       value += static_cast<int64_t>(consumed) * delta;
     } else {
       if (notNull) {
         for (uint64_t i = 0 ; i < count; ++i) {
           if (notNull[position + i]) {
             data[position + i] = isSigned
                 ? unZigZag(readLong())
                 : static_cast<int64_t>(readLong());
             ++consumed;
           }
         }
       } else {
         if (isSigned) {
           for (uint64_t i = 0; i < count; ++i) {
             data[position + i] = unZigZag(readLong());
           }
         } else {
           for (uint64_t i = 0; i < count; ++i) {
             data[position + i] = static_cast<int64_t>(readLong());
           }
         }
         consumed = count;
       }
     }
     remainingValues -= consumed;
     position += count;

     // skipNulls()
     if (notNull) {
       // Skip over null values.
       while (position < numValues && !notNull[position]) {
         ++position;
       }
     }
   }
 }

 }  // namespace orc
	/**
	* 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 "orc/Exceptions.hh"
	#include "RLEv1.hh"

	#include <algorithm>

	namespace orc {

	const uint64_t MINIMUM_REPEAT = 3;
	const uint64_t MAXIMUM_REPEAT = 127 + MINIMUM_REPEAT;

	const int64_t BASE_128_MASK = 0x7f;

	const int64_t MAX_DELTA = 127;
	const int64_t MIN_DELTA = -128;
	const uint64_t MAX_LITERAL_SIZE = 128;

	RleEncoderV1::RleEncoderV1(
	std::unique_ptr<BufferedOutputStream> outStream,
	bool hasSigned):
	RleEncoder(std::move(outStream), hasSigned) {
	literals = new int64_t[MAX_LITERAL_SIZE];
	delta = 0;
	repeat = false;
	tailRunLength = 0;
	}

	RleEncoderV1::~RleEncoderV1() {
	delete [] literals;
	}

	void RleEncoderV1::writeValues() {
	if (numLiterals != 0) {
	if (repeat) {
	writeByte(static_cast<char>
	(static_cast<uint64_t>(numLiterals) - MINIMUM_REPEAT));
	writeByte(static_cast<char>(delta));
	if (isSigned) {
	writeVslong(literals[0]);
	} else {
	writeVulong(literals[0]);
	}
	} else {
	writeByte(static_cast<char>(-numLiterals));
	for(size_t i=0; i < numLiterals; ++i) {
	if (isSigned) {
	writeVslong(literals[i]);
	} else {
	writeVulong(literals[i]);
	}
	}
	}
	repeat = false;
	numLiterals = 0;
	tailRunLength = 0;
	}
	}

	uint64_t RleEncoderV1::flush() {
	writeValues();
	outputStream->BackUp(static_cast<int>(bufferLength - bufferPosition));
	uint64_t dataSize = outputStream->flush();
	bufferLength = bufferPosition = 0;
	return dataSize;
	}

	void RleEncoderV1::write(int64_t value) {
	if (numLiterals == 0) {
	literals[numLiterals++] = value;
	tailRunLength = 1;
	} else if (repeat) {
	if (value == literals[0] + delta * static_cast<int64_t>(numLiterals)) {
	numLiterals += 1;
	if (numLiterals == MAXIMUM_REPEAT) {
	writeValues();
	}
	} else {
	writeValues();
	literals[numLiterals++] = value;
	tailRunLength = 1;
	}
	} else {
	if (tailRunLength == 1) {
	delta = value - literals[numLiterals - 1];
	if (delta < MIN_DELTA \|\| delta > MAX_DELTA) {
	tailRunLength = 1;
	} else {
	tailRunLength = 2;
	}
	} else if (value == literals[numLiterals - 1] + delta) {
	tailRunLength += 1;
	} else {
	delta = value - literals[numLiterals - 1];
	if (delta < MIN_DELTA \|\| delta > MAX_DELTA) {
	tailRunLength = 1;
	} else {
	tailRunLength = 2;
	}
	}
	if (tailRunLength == MINIMUM_REPEAT) {
	if (numLiterals + 1 == MINIMUM_REPEAT) {
	repeat = true;
	numLiterals += 1;
	} else {
	numLiterals -= static_cast<int>(MINIMUM_REPEAT - 1);
	int64_t base = literals[numLiterals];
	writeValues();
	literals[0] = base;
	repeat = true;
	numLiterals = MINIMUM_REPEAT;
	}
	} else {
	literals[numLiterals++] = value;
	if (numLiterals == MAX_LITERAL_SIZE) {
	writeValues();
	}
	}
	}
	}

	signed char RleDecoderV1::readByte() {
	if (bufferStart == bufferEnd) {
	int bufferLength;
	const void* bufferPointer;
	if (!inputStream->Next(&bufferPointer, &bufferLength)) {
	throw ParseError("bad read in readByte");
	}
	bufferStart = static_cast<const char*>(bufferPointer);
	bufferEnd = bufferStart + bufferLength;
	}
	return *(bufferStart++);
	}

	uint64_t RleDecoderV1::readLong() {
	uint64_t result = 0;
	int64_t offset = 0;
	signed char ch = readByte();
	if (ch >= 0) {
	result = static_cast<uint64_t>(ch);
	} else {
	result = static_cast<uint64_t>(ch) & BASE_128_MASK;
	while ((ch = readByte()) < 0) {
	offset += 7;
	result \|= (static_cast<uint64_t>(ch) & BASE_128_MASK) << offset;
	}
	result \|= static_cast<uint64_t>(ch) << (offset + 7);
	}
	return result;
	}

	void RleDecoderV1::skipLongs(uint64_t numValues) {
	while (numValues > 0) {
	if (readByte() >= 0) {
	--numValues;
	}
	}
	}

	void RleDecoderV1::readHeader() {
	signed char ch = readByte();
	if (ch < 0) {
	remainingValues = static_cast<uint64_t>(-ch);
	repeating = false;
	} else {
	remainingValues = static_cast<uint64_t>(ch) + MINIMUM_REPEAT;
	repeating = true;
	delta = readByte();
	value = isSigned
	? unZigZag(readLong())
	: static_cast<int64_t>(readLong());
	}
	}

	RleDecoderV1::RleDecoderV1(std::unique_ptr<SeekableInputStream> input,
	bool hasSigned)
	: inputStream(std::move(input)),
	isSigned(hasSigned),
	remainingValues(0),
	value(0),
	bufferStart(nullptr),
	bufferEnd(bufferStart),
	delta(0),
	repeating(false) {
	}

	void RleDecoderV1::seek(PositionProvider& location) {
	// move the input stream
	inputStream->seek(location);
	// force a re-read from the stream
	bufferEnd = bufferStart;
	// read a new header
	readHeader();
	// skip ahead the given number of records
	skip(location.next());
	}

	void RleDecoderV1::skip(uint64_t numValues) {
	while (numValues > 0) {
	if (remainingValues == 0) {
	readHeader();
	}
	uint64_t count = std::min(numValues, remainingValues);
	remainingValues -= count;
	numValues -= count;
	if (repeating) {
	value += delta * static_cast<int64_t>(count);
	} else {
	skipLongs(count);
	}
	}
	}

	void RleDecoderV1::next(int64_t* const data,
	const uint64_t numValues,
	const char* const notNull) {
	uint64_t position = 0;
	// skipNulls()
	if (notNull) {
	// Skip over null values.
	while (position < numValues && !notNull[position]) {
	++position;
	}
	}
	while (position < numValues) {
	// If we are out of values, read more.
	if (remainingValues == 0) {
	readHeader();
	}
	// How many do we read out of this block?
	uint64_t count = std::min(numValues - position, remainingValues);
	uint64_t consumed = 0;
	if (repeating) {
	if (notNull) {
	for (uint64_t i = 0; i < count; ++i) {
	if (notNull[position + i]) {
	data[position + i] = value + static_cast<int64_t>(consumed) * delta;
	consumed += 1;
	}
	}
	} else {
	for (uint64_t i = 0; i < count; ++i) {
	data[position + i] = value + static_cast<int64_t>(i) * delta;
	}
	consumed = count;
	}
	value += static_cast<int64_t>(consumed) * delta;
	} else {
	if (notNull) {
	for (uint64_t i = 0 ; i < count; ++i) {
	if (notNull[position + i]) {
	data[position + i] = isSigned
	? unZigZag(readLong())
	: static_cast<int64_t>(readLong());
	++consumed;
	}
	}
	} else {
	if (isSigned) {
	for (uint64_t i = 0; i < count; ++i) {
	data[position + i] = unZigZag(readLong());
	}
	} else {
	for (uint64_t i = 0; i < count; ++i) {
	data[position + i] = static_cast<int64_t>(readLong());
	}
	}
	consumed = count;
	}
	}
	remainingValues -= consumed;
	position += count;

	// skipNulls()
	if (notNull) {
	// Skip over null values.
	while (position < numValues && !notNull[position]) {
	++position;
	}
	}
	}
	}

	} // namespace orc