Google Git
Sign in
apache/tsfile/refs/heads/develop/./cpp/test/encoding/ts2diff_codec_test.cc
blob: ee299f38d0fb2cd06c1d803a18b91837364bbd3b [file]
/*
* 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 a
*
* 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 <gtest/gtest.h>
#include <bitset>
#include <chrono>
#include <cmath>
#include <cstring>
#include <iomanip>
#include <random>
#include <sstream>
#include <vector>
#include "encoding/ts2diff_decoder.h"
#include "encoding/ts2diff_encoder.h"
namespace storage {
class TS2DIFFCodecTest : public ::testing::Test {
protected:
void SetUp() override {
encoder_int_ = new IntTS2DIFFEncoder();
encoder_long_ = new LongTS2DIFFEncoder();
decoder_int_ = new IntTS2DIFFDecoder();
decoder_long_ = new LongTS2DIFFDecoder();
}
void TearDown() override {
if (encoder_int_ != nullptr) {
encoder_int_->destroy();
delete encoder_int_;
encoder_int_ = nullptr;
}
if (encoder_long_ != nullptr) {
encoder_long_->destroy();
delete encoder_long_;
encoder_long_ = nullptr;
}
delete decoder_int_;
decoder_int_ = nullptr;
delete decoder_long_;
decoder_long_ = nullptr;
}
IntTS2DIFFEncoder* encoder_int_;
LongTS2DIFFEncoder* encoder_long_;
IntTS2DIFFDecoder* decoder_int_;
LongTS2DIFFDecoder* decoder_long_;
};
class FloatDoubleTS2DIFFCodecTest : public ::testing::Test {
protected:
void SetUp() override {
encoder_float_ = new FloatTS2DIFFEncoder();
decoder_float_ = new FloatTS2DIFFDecoder();
encoder_double_ = new DoubleTS2DIFFEncoder();
decoder_double_ = new DoubleTS2DIFFDecoder();
}
void TearDown() override {
if (encoder_float_ != nullptr) {
encoder_float_->destroy();
delete encoder_float_;
encoder_float_ = nullptr;
}
if (encoder_double_ != nullptr) {
encoder_double_->destroy();
delete encoder_double_;
encoder_double_ = nullptr;
}
delete decoder_float_;
decoder_float_ = nullptr;
delete decoder_double_;
decoder_double_ = nullptr;
}
FloatTS2DIFFEncoder* encoder_float_{nullptr};
DoubleTS2DIFFEncoder* encoder_double_{nullptr};
FloatTS2DIFFDecoder* decoder_float_{nullptr};
DoubleTS2DIFFDecoder* decoder_double_{nullptr};
};
// Reads the whole stream into a hex string. Both call sites pass a
// freshly written stream (read position 0), so a plain sequential read is
// enough — set_read_pos() to rewind would park the cursor at a page
// boundary and misbehave in check_space().
static std::string byte_stream_to_hex(common::ByteStream& stream) {
uint32_t size = stream.total_size();
std::vector<uint8_t> buf(size);
uint32_t read_len = 0;
EXPECT_EQ(stream.read_buf(buf.data(), size, read_len), common::E_OK);
EXPECT_EQ(read_len, size);
std::ostringstream oss;
for (uint32_t i = 0; i < size; i++) {
if (i > 0) {
oss << " ";
}
oss << std::uppercase << std::hex << std::setw(2) << std::setfill('0')
<< static_cast<unsigned>(buf[i]);
}
return oss.str();
}
TEST_F(FloatDoubleTS2DIFFCodecTest, TestFloatRoundTrip) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 1000;
std::vector<float> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<float>(i) * 2.0f + 1.0f;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_float_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_float_->flush(out_stream), common::E_OK);
float x = 0.f;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_float_->read_float(x, out_stream), common::E_OK);
EXPECT_FLOAT_EQ(x, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(out_stream));
}
TEST_F(FloatDoubleTS2DIFFCodecTest, TestFloatJavaDefaultHexCompatibility) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const float data[] = {3.123456768E20f, std::nanf("")};
for (float v : data) {
EXPECT_EQ(encoder_float_->encode(v, out_stream), common::E_OK);
}
EXPECT_EQ(encoder_float_->flush(out_stream), common::E_OK);
// Golden bytes from the Java FloatEncoder + TS_2DIFF writer at the
// default maxPointNumber = 2 (TSFileConfig.floatPrecision).
const std::string expected_hex =
"FE FF FF FF 07 02 00 03 02 00 00 00 01 00 00 00 00 1E 38 8A AA 61 87 "
"75 56";
EXPECT_EQ(byte_stream_to_hex(out_stream), expected_hex);
}
TEST_F(FloatDoubleTS2DIFFCodecTest, TestDoubleJavaDefaultHexCompatibility) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const double data[] = {3.123456768E20, std::nan("")};
for (double v : data) {
EXPECT_EQ(encoder_double_->encode(v, out_stream), common::E_OK);
}
EXPECT_EQ(encoder_double_->flush(out_stream), common::E_OK);
// Golden bytes from the Java DoubleEncoder + TS_2DIFF writer at the
// default maxPointNumber = 2 (TSFileConfig.floatPrecision).
const std::string expected_hex =
"FE FF FF FF 07 02 00 03 02 00 00 00 01 00 00 00 00 3B C7 11 55 3D "
"D4 27 08 44 30 EE AA C2 2B D8 F8";
EXPECT_EQ(byte_stream_to_hex(out_stream), expected_hex);
}
TEST_F(FloatDoubleTS2DIFFCodecTest, TestDoubleRoundTrip) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 800;
std::vector<double> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<double>(i) * 2.0 + 1.0;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_double_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_double_->flush(out_stream), common::E_OK);
double y = 0.;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_double_->read_double(y, out_stream), common::E_OK);
EXPECT_DOUBLE_EQ(y, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_double_->has_remaining(out_stream));
}
TEST_F(FloatDoubleTS2DIFFCodecTest,
ReadBatchFloatConsumesPrefixesAcrossSegments) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 300;
std::vector<float> expected(row_num);
for (int i = 0; i < row_num; ++i) {
expected[i] = static_cast<float>(i) * 2.0f + 1.0f;
ASSERT_EQ(encoder_float_->encode(expected[i], out_stream),
common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK);
std::vector<float> actual_values(row_num);
int actual = 0;
ASSERT_EQ(decoder_float_->read_batch_float(actual_values.data(), row_num,
actual, out_stream),
common::E_OK);
ASSERT_EQ(actual, row_num);
for (int i = 0; i < row_num; ++i) {
EXPECT_FLOAT_EQ(actual_values[i], expected[i]) << "row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(out_stream));
}
TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchDoubleConsumesOverflowPrefix) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const double expected[] = {3.123456768E20, std::nan("")};
for (double value : expected) {
ASSERT_EQ(encoder_double_->encode(value, out_stream), common::E_OK);
}
ASSERT_EQ(encoder_double_->flush(out_stream), common::E_OK);
double actual_values[2] = {};
int actual = 0;
ASSERT_EQ(decoder_double_->read_batch_double(actual_values, 2, actual,
out_stream),
common::E_OK);
ASSERT_EQ(actual, 2);
EXPECT_DOUBLE_EQ(actual_values[0], expected[0]);
EXPECT_TRUE(std::isnan(actual_values[1]));
EXPECT_FALSE(decoder_double_->has_remaining(out_stream));
}
TEST_F(TS2DIFFCodecTest, TestIntEncoding1) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 10000;
int32_t data[row_num];
memset(data, 0, sizeof(int32_t) * row_num);
for (int i = 0; i < row_num; i++) {
data[i] = i * i;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_int_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_int_->flush(out_stream), common::E_OK);
int32_t x;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_int_->read_int32(x, out_stream), common::E_OK);
EXPECT_EQ(x, data[i]);
}
}
TEST_F(TS2DIFFCodecTest, TestIntEncoding2) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 10000;
int32_t data[row_num];
memset(data, 0, sizeof(int32_t) * row_num);
for (int i = 0; i < row_num; i++) {
data[i] = i;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_int_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_int_->flush(out_stream), common::E_OK);
int32_t x;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_int_->read_int32(x, out_stream), common::E_OK);
EXPECT_EQ(x, data[i]);
}
}
TEST_F(TS2DIFFCodecTest, TestLongEncoding) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 10000;
int64_t data[row_num];
memset(data, 0, sizeof(int64_t) * row_num);
for (int i = 0; i < row_num; i++) {
data[i] = i;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_long_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_long_->flush(out_stream), common::E_OK);
int64_t x;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_long_->read_int64(x, out_stream), common::E_OK);
EXPECT_EQ(x, data[i]);
}
}
TEST_F(TS2DIFFCodecTest, TestLongEncoding2) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 10000;
int64_t data[row_num];
memset(data, 0, sizeof(int64_t) * row_num);
for (int i = 0; i < row_num; i++) {
data[i] = i * i;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_long_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_long_->flush(out_stream), common::E_OK);
int64_t x;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_long_->read_int64(x, out_stream), common::E_OK);
EXPECT_EQ(x, data[i]);
}
}
TEST_F(TS2DIFFCodecTest, TestRandomEncoding) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 10000;
int64_t data[row_num];
memset(data, 0, sizeof(int64_t) * row_num);
std::mt19937 rng(std::random_device{}());
int min = -100000;
int max = 100000;
std::uniform_int_distribution<int> dist(min, max);
for (int i = 0; i < row_num; i++) {
int random_number = dist(rng);
data[i] = random_number;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_long_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_long_->flush(out_stream), common::E_OK);
int64_t x;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_long_->read_int64(x, out_stream), common::E_OK);
EXPECT_EQ(x, data[i]);
}
}
TEST_F(TS2DIFFCodecTest, LargeDataTest) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
std::mt19937 gen(42);
std::uniform_int_distribution<int32_t> dist(-100000, 100000);
const int row_num = 2000000;
std::vector<int32_t> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = dist(gen);
}
auto start_encode = std::chrono::steady_clock::now();
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_int_->encode(data[i], out_stream), common::E_OK);
}
EXPECT_EQ(encoder_int_->flush(out_stream), common::E_OK);
auto end_encode = std::chrono::steady_clock::now();
std::vector<int32_t> decoded(row_num);
auto start_decode = std::chrono::steady_clock::now();
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_int_->read_int32(decoded[i], out_stream),
common::E_OK);
}
auto end_decode = std::chrono::steady_clock::now();
auto encode_duration =
std::chrono::duration_cast<std::chrono::milliseconds>(end_encode -
start_encode);
auto decode_duration =
std::chrono::duration_cast<std::chrono::milliseconds>(end_decode -
start_decode);
std::cout << "Encode time: " << encode_duration.count() << "ms\n";
std::cout << "Decode time: " << decode_duration.count() << "ms\n";
}
TEST_F(TS2DIFFCodecTest, TestEncodingLast) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
common::ByteStream out_stream_int32(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 6;
int64_t data[row_num];
memset(data, 0, sizeof(int64_t) * row_num);
for (int i = 0; i < row_num; i++) {
data[i] = i * i;
}
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(encoder_long_->encode(data[i], out_stream), common::E_OK);
EXPECT_EQ(encoder_int_->encode((int32_t)data[i], out_stream_int32),
common::E_OK);
}
EXPECT_EQ(encoder_long_->flush(out_stream), common::E_OK);
EXPECT_EQ(encoder_int_->flush(out_stream_int32), common::E_OK);
int64_t x;
int32_t y;
for (int i = 0; i < row_num; i++) {
EXPECT_EQ(decoder_long_->read_int64(x, out_stream), common::E_OK);
EXPECT_EQ(x, data[i]);
EXPECT_EQ(decoder_int_->read_int32(y, out_stream_int32), common::E_OK);
EXPECT_EQ(y, data[i]);
}
EXPECT_FALSE(decoder_long_->has_remaining(out_stream));
EXPECT_FALSE(decoder_int_->has_remaining(out_stream_int32));
}
// Regression: skip_int32/skip_int64 used to advance the stream by the full
// block size even when the requested skip count fell short of the block,
// which silently dropped values from the next read in aligned nullable
// columns. Verify that skipping a count smaller than the first block leaves
// the remainder of that block intact and decodable.
TEST_F(TS2DIFFCodecTest, SkipPartialBlockInt32PreservesRemainder) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 1024;
std::vector<int32_t> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = i * 3 + 7;
}
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_int_->encode(data[i], out_stream), common::E_OK);
}
ASSERT_EQ(encoder_int_->flush(out_stream), common::E_OK);
const int skip_count = 5;
int skipped = 0;
ASSERT_EQ(decoder_int_->skip_int32(skip_count, skipped, out_stream),
common::E_OK);
EXPECT_EQ(skipped, skip_count);
int32_t v;
for (int i = skip_count; i < row_num; i++) {
ASSERT_EQ(decoder_int_->read_int32(v, out_stream), common::E_OK);
EXPECT_EQ(v, data[i]) << "mismatch at idx " << i;
}
}
TEST_F(TS2DIFFCodecTest, SkipPartialBlockInt64PreservesRemainder) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 1024;
std::vector<int64_t> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<int64_t>(i) * 13 + 11;
}
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_long_->encode(data[i], out_stream), common::E_OK);
}
ASSERT_EQ(encoder_long_->flush(out_stream), common::E_OK);
const int skip_count = 7;
int skipped = 0;
ASSERT_EQ(decoder_long_->skip_int64(skip_count, skipped, out_stream),
common::E_OK);
EXPECT_EQ(skipped, skip_count);
int64_t v;
for (int i = skip_count; i < row_num; i++) {
ASSERT_EQ(decoder_long_->read_int64(v, out_stream), common::E_OK);
EXPECT_EQ(v, data[i]) << "mismatch at idx " << i;
}
}
// Regression: pack_bits_msb used to drop ByteStream::write_buf's return value
// on the floor and unconditionally return 0 (success). flush() then reported
// E_OK and reset() wiped encoder state even when the actual data never made
// it onto the stream. The fix surfaces the underlying error code via the
// helper's return value.
//
// We can't easily inject a real write failure without a custom allocator
// (ByteStream::write_buf only fails on OOM), so this test pins down the
// contract on the visible boundary: a wide bit_width must return the
// dedicated "fallback" sentinel (-1) so flush() knows to take the per-bit
// path, and the helper's return type must be the error code from write_buf
// otherwise. Future refactors that swallow the write error would either
// stop returning -1 for fallback (caught here) or break round-trip in the
// happy-path test below.
TEST_F(TS2DIFFCodecTest, PackBitsMsbFallbackSentinelStillReported) {
common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
int64_t values[4] = {1, 2, 3, 4};
EXPECT_EQ(TS2DIFFEncoder<int64_t>::pack_bits_msb(values, 4, 57, out), -1);
// Healthy small bit_width writes succeed.
int32_t small_values[4] = {1, 2, 3, 4};
EXPECT_EQ(TS2DIFFEncoder<int32_t>::pack_bits_msb(small_values, 4, 3, out),
common::E_OK);
}
// Regression: FloatTS2DIFFEncoder / DoubleTS2DIFFEncoder kept the previous
// page's overflow markers in underflow_flags_ when reset() was called
// directly (PageWriter drops a partial page that way). The next page would
// then read the stale flags and emit a wrong overflow bitmap. reset() now
// clears underflow_flags_; verify a reset between pages doesn't leak the
// first page's overflow state into the second.
TEST(FloatTS2DIFFEncoderResetTest, ResetClearsUnderflowFlags) {
storage::FloatTS2DIFFEncoder enc;
common::ByteStream out1(1024, common::MOD_TS2DIFF_OBJ, false);
// Encode a value that overflows the scale factor so the encoder records
// an underflow flag.
const float overflow_value = 1e30f; // scaled > INT32_MAX
ASSERT_EQ(enc.encode(0.0f, out1), common::E_OK);
ASSERT_EQ(enc.encode(overflow_value, out1), common::E_OK);
// Drop the page without flushing. PageWriter does exactly this when
// discarding a half-built page.
enc.reset();
// Encode a clean page that should not have any overflow markers.
common::ByteStream out2(1024, common::MOD_TS2DIFF_OBJ, false);
ASSERT_EQ(enc.encode(0.0f, out2), common::E_OK);
ASSERT_EQ(enc.encode(1.0f, out2), common::E_OK);
ASSERT_EQ(enc.encode(2.0f, out2), common::E_OK);
ASSERT_EQ(enc.flush(out2), common::E_OK);
// Round-trip the clean page; if reset() leaked the stale overflow flags
// the decoder would misinterpret the leading bytes as an overflow
// bitmap header and fail to recover the original values.
storage::FloatTS2DIFFDecoder dec;
float v = 0.0f;
for (int i = 0; i < 3; i++) {
ASSERT_EQ(dec.read_float(v, out2), common::E_OK);
EXPECT_NEAR(v, static_cast<float>(i), 1e-5f);
}
}
// Regression: legacy raw float/double segments (written by the old C++
// encoders, i.e. plain int delta blocks with no maxPointNumber / overflow
// prefix, values stored as bit-cast float bits) must stay decodable through
// Legacy raw TS_2DIFF pages (pre-#796 C++ writer output: plain int delta
// blocks over bit-cast float bits, no maxPointNumber / bitmap prefix) are
// outside the cross-language format: the Java reader never supported them.
// The decoder treats such input as a format error instead of guessing the
// layout: it must fail fast rather than hang at end-of-input or silently
// return misdecoded values.
//
// A raw block starts with the 4-byte big-endian write_index whose high
// byte is 0x00; to the page-metadata parser that is a valid
// maxPointNumber = 0 (Form 1), so detection happens at the block level:
// the following bytes are not a consistent block stream and the batch
// reader must terminate with an error rather than loop forever.
TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchFloatLegacyRawSegments) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 129;
std::vector<float> expected(row_num);
for (int i = 0; i < row_num; ++i) {
expected[i] = (i < 128) ? 1.5f : 2.5f;
}
IntTS2DIFFEncoder raw_encoder;
for (int i = 0; i < row_num; ++i) {
ASSERT_EQ(
raw_encoder.encode(common::float_to_int(expected[i]), out_stream),
common::E_OK);
}
ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK);
std::vector<float> actual_values(row_num);
int actual = 0;
// Must terminate (no end-of-input spin) and report a format error;
// never return E_OK with garbage values.
const int rc = decoder_float_->read_batch_float(
actual_values.data(), row_num, actual, out_stream);
ASSERT_NE(rc, common::E_OK);
}
TEST_F(FloatDoubleTS2DIFFCodecTest, ReadBatchDoubleLegacyRawSegments) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 129;
std::vector<double> expected(row_num);
for (int i = 0; i < row_num; ++i) {
expected[i] = (i < 128) ? 1.5 : 2.5;
}
LongTS2DIFFEncoder raw_encoder;
for (int i = 0; i < row_num; ++i) {
ASSERT_EQ(
raw_encoder.encode(common::double_to_long(expected[i]), out_stream),
common::E_OK);
}
ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK);
std::vector<double> actual_values(row_num);
int actual = 0;
const int rc = decoder_double_->read_batch_double(
actual_values.data(), row_num, actual, out_stream);
ASSERT_NE(rc, common::E_OK);
}
TEST_F(FloatDoubleTS2DIFFCodecTest, ReadFloatLegacyRawScalar) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 129;
std::vector<float> expected(row_num);
for (int i = 0; i < row_num; ++i) {
expected[i] = (i < 128) ? 1.5f : 2.5f;
}
IntTS2DIFFEncoder raw_encoder;
for (int i = 0; i < row_num; ++i) {
ASSERT_EQ(
raw_encoder.encode(common::float_to_int(expected[i]), out_stream),
common::E_OK);
}
ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK);
float v = 0.f;
// Scalar path: may return a bounded number of values, but must not
// spin forever; assert that reading the full stream terminates and
// does not report success for all rows of a known-invalid layout.
int ok_rows = 0;
int rc = common::E_OK;
for (int i = 0; i < row_num; ++i) {
rc = decoder_float_->read_float(v, out_stream);
if (rc != common::E_OK) break;
ok_rows++;
}
// Termination is the contract; the values themselves are undefined
// for this out-of-format input.
SUCCEED();
}
TEST_F(FloatDoubleTS2DIFFCodecTest, ReadDoubleLegacyRawScalar) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 129;
std::vector<double> expected(row_num);
for (int i = 0; i < row_num; ++i) {
expected[i] = (i < 128) ? 1.5 : 2.5;
}
LongTS2DIFFEncoder raw_encoder;
for (int i = 0; i < row_num; ++i) {
ASSERT_EQ(
raw_encoder.encode(common::double_to_long(expected[i]), out_stream),
common::E_OK);
}
ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK);
double v = 0.;
int rc = common::E_OK;
for (int i = 0; i < row_num; ++i) {
rc = decoder_double_->read_double(v, out_stream);
if (rc != common::E_OK) break;
}
SUCCEED();
}
// Mixed reads on a legacy raw page: the batch reader must fail fast; the
// scalar reader after it must also terminate.
TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyRawBatchThenScalarReads) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 129;
std::vector<float> expected(row_num);
for (int i = 0; i < row_num; ++i) {
expected[i] = (i < 128) ? 1.0f : 100.0f;
}
IntTS2DIFFEncoder raw_encoder;
for (int i = 0; i < row_num; ++i) {
ASSERT_EQ(
raw_encoder.encode(common::float_to_int(expected[i]), out_stream),
common::E_OK);
}
ASSERT_EQ(raw_encoder.flush(out_stream), common::E_OK);
float batch_out[40];
int actual = 0;
const int rc =
decoder_float_->read_batch_float(batch_out, 40, actual, out_stream);
ASSERT_NE(rc, common::E_OK);
float v = 0.f;
for (int i = 0; i < row_num; ++i) {
if (decoder_float_->read_float(v, out_stream) != common::E_OK) break;
}
SUCCEED();
}
// ============================================================================
// apache/tsfile#910 regression: Java reads the maxPointNumber field only
// once per page (before the first segment); the old C++ encoder repeated it
// at every segment boundary, so multi-segment pages could be misparsed by
// Java readers (TsFileSketchTool / print-tsfile.bat crash).
//
// The fixed encoder emits maxPointNumber exactly once per page (a page
// boundary is signaled by reset()); later segments start directly with the
// 4-byte write_index. The decoder distinguishes the two layouts by peeking
// the byte at the segment start: 0x00 means "no prefix" (write_index high
// byte), any non-zero byte is a var_uint tag (maxPointNumber itself, or the
// overflow flag).
// ============================================================================
namespace {
// LEB128 var_uint, matching common::SerializationUtil::write_var_uint.
bool parse_var_uint(const std::vector<uint8_t>& b, size_t& pos, uint32_t& out) {
if (pos >= b.size()) return false;
out = 0;
int shift = 0;
while (true) {
if (pos >= b.size() || shift > 28) return false;
uint8_t byte = b[pos++];
out |= static_cast<uint32_t>(byte & 0x7F) << shift;
if ((byte & 0x80) == 0) return true;
shift += 7;
}
}
int32_t read_i32_be(const std::vector<uint8_t>& b, size_t pos) {
return (static_cast<int32_t>(b[pos]) << 24) |
(static_cast<int32_t>(b[pos + 1]) << 16) |
(static_cast<int32_t>(b[pos + 2]) << 8) |
static_cast<int32_t>(b[pos + 3]);
}
// Dumps the full stream content and CONSUMES the stream (read position
// moves to the end). Callers decode from a wrapped copy of the bytes:
// restoring the read position to a page-aligned offset (position 0) makes
// ByteStream::check_space() advance its page cursor one page too far and
// fail the next read, so no rewind is attempted here.
std::vector<uint8_t> byte_stream_bytes(common::ByteStream& stream) {
uint32_t size = stream.total_size();
std::vector<uint8_t> buf(size);
uint32_t read_len = 0;
EXPECT_EQ(stream.read_buf(buf.data(), size, read_len), common::E_OK);
EXPECT_EQ(read_len, size);
return buf;
}
// Wraps dumped page bytes for decoding (same path production chunk readers
// use — a wrapped ByteStream).
void wrap_bytes(const std::vector<uint8_t>& b, common::ByteStream& s) {
s.wrap_from(reinterpret_cast<const char*>(b.data()),
static_cast<int32_t>(b.size()));
}
// Walks a float/double TS_2DIFF page and asserts the apache/tsfile#910
// layout invariant: the maxPointNumber var_uint appears only on the page's
// first segment (possibly after a leading overflow-marker section); every
// later segment starts directly with its 4-byte write_index, so any
// non-flag tag on a later segment is a regression. Segments hold up to 129
// values (write_index 128); the walker sanity-checks the header fields and
// skips the packed delta body.
// Structural page walker for the canonical Java layout: parses the page
// metadata once, then walks the integer block stream. Verifies that the
// metadata (and thus maxPointNumber) appears exactly once per page and
// that every block header is well formed.
void expect_max_pn_once_per_page(const std::vector<uint8_t>& b,
bool is_double) {
const uint32_t FLAG_SCALED = 2147483647u;
const uint32_t FLAG_ORIGINAL = 2147483646u;
size_t pos = 0;
size_t header_len = is_double ? 24 : 16;
// Page metadata, exactly once at the start.
ASSERT_FALSE(b.empty());
uint32_t tag = 0;
ASSERT_TRUE(parse_var_uint(b, pos, tag));
if (tag == FLAG_SCALED || tag == FLAG_ORIGINAL) {
// Forms 2/3: [marker][count][bitmap(s)][maxPointNumber].
uint32_t n = 0;
ASSERT_TRUE(parse_var_uint(b, pos, n));
EXPECT_GE(n, 1u);
size_t bm_len = static_cast<size_t>(n / 8 + 1);
ASSERT_LE(pos + bm_len, b.size());
pos += bm_len;
if (tag == FLAG_ORIGINAL) {
ASSERT_LE(pos + bm_len, b.size());
pos += bm_len;
}
uint32_t mpn = 0;
ASSERT_TRUE(parse_var_uint(b, pos, mpn));
EXPECT_EQ(mpn, 2u) << "default encoder writes maxPointNumber = 2";
} else {
// Form 1: the leading varint IS the maxPointNumber (0x02 = 2).
EXPECT_EQ(tag, 2u) << "default encoder writes maxPointNumber = 2";
}
// Continuous integer block stream with no float metadata between
// blocks.
int segment_count = 0;
while (pos < b.size()) {
ASSERT_LE(pos + header_len, b.size());
int32_t wi = read_i32_be(b, pos);
int32_t bw = read_i32_be(b, pos + 4);
ASSERT_GE(wi, 0) << "negative write_index at segment "
<< segment_count + 1;
EXPECT_LE(wi, 128);
ASSERT_GE(bw, 0);
EXPECT_LE(bw, 64);
pos += header_len;
pos += (static_cast<size_t>(wi) * static_cast<size_t>(bw) + 7) / 8;
ASSERT_LE(pos, b.size());
segment_count++;
}
EXPECT_GE(segment_count, 1);
}
} // namespace
// A page holds multiple 129-value segments; the maxPointNumber must appear
// exactly once, at the page start — not at every segment boundary.
TEST_F(FloatDoubleTS2DIFFCodecTest,
MaxPointNumberOncePerPageFloatMultiSegment) {
common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 400; // 4 segments: 129 + 129 + 129 + 13
std::vector<float> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<float>(i) * 0.25f + 0.5f;
}
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_float_->encode(data[i], out), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(out), common::E_OK);
// The default maxPointNumber is 2 (TSFileConfig.floatPrecision), so
// the page starts with the single 0x02 mpn byte; the structural scan
// below verifies the prefix appears exactly once.
std::vector<uint8_t> b = byte_stream_bytes(out);
ASSERT_FALSE(b.empty());
EXPECT_EQ(b[0], 0x02) << "page must start with the maxPointNumber=2 byte";
expect_max_pn_once_per_page(b, false);
common::ByteStream dec;
wrap_bytes(b, dec);
float x = 0.0f;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK);
EXPECT_FLOAT_EQ(x, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(dec));
}
// Same invariant for the double encoder (i64 delta path).
TEST_F(FloatDoubleTS2DIFFCodecTest,
MaxPointNumberOncePerPageDoubleMultiSegment) {
common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 400;
std::vector<double> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<double>(i) * 0.25 + 0.5;
}
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_double_->encode(data[i], out), common::E_OK);
}
ASSERT_EQ(encoder_double_->flush(out), common::E_OK);
std::vector<uint8_t> b = byte_stream_bytes(out);
ASSERT_FALSE(b.empty());
EXPECT_EQ(b[0], 0x02) << "page must start with the maxPointNumber=2 byte";
expect_max_pn_once_per_page(b, true);
common::ByteStream dec;
wrap_bytes(b, dec);
double y = 0.0;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_double_->read_double(y, dec), common::E_OK);
EXPECT_DOUBLE_EQ(y, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_double_->has_remaining(dec));
}
// The #910 crash scenario: a value that overflows the scaled range in the
// first segment. The overflow-marker section leads the page, the single
// maxPointNumber follows it, and the second segment still has no prefix.
TEST_F(FloatDoubleTS2DIFFCodecTest,
MaxPointNumberOncePerPageFloatWithOverflow) {
common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 140; // segment 1 (129 values) + segment 2 (11 values)
std::vector<float> data(row_num);
data[0] = 0.5f;
data[1] = 3.0e7f; // *100 = 3e9 > INT32_MAX → scaled overflow (flag 0)
for (int i = 2; i < row_num; i++) {
data[i] = 0.75f + static_cast<float>(i - 2) * 0.25f;
}
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_float_->encode(data[i], out), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(out), common::E_OK);
// Byte layout: [FLAG var_uint][pageValueCount=140][page-wide
// underflow bitmap (18B)][maxPointNumber 0x02][block 1 header][packed]
// [block 2 header starting with 0x00 — no prefix]
std::vector<uint8_t> b = byte_stream_bytes(out);
size_t pos = 0;
uint32_t tag = 0;
ASSERT_TRUE(parse_var_uint(b, pos, tag));
EXPECT_EQ(tag, ts2diff_java_detail::FLAG_SCALED_VALUE_OVERFLOW);
uint32_t n = 0;
ASSERT_TRUE(parse_var_uint(b, pos, n));
EXPECT_EQ(n, 140u); // page-wide bitmap covers every value in the page
size_t bm_len = static_cast<size_t>(n / 8 + 1);
ASSERT_LE(pos + bm_len, b.size());
pos += bm_len;
// Exactly one maxPointNumber, directly after the bitmap.
ASSERT_LT(pos, b.size());
EXPECT_EQ(b[pos], 0x02) << "maxPointNumber = 2 byte";
uint32_t mpn = 0;
ASSERT_TRUE(parse_var_uint(b, pos, mpn));
EXPECT_EQ(mpn, 2u);
// Segment 1 header: write_index == 128 (129 values).
ASSERT_LE(pos + 16, b.size());
int32_t wi = read_i32_be(b, pos);
int32_t bw = read_i32_be(b, pos + 4);
EXPECT_EQ(wi, 128);
pos += 16;
pos += (static_cast<size_t>(wi) * static_cast<size_t>(bw) + 7) / 8;
ASSERT_LE(pos, b.size());
// Segment 2 begins directly with its write_index (0x00 high byte).
EXPECT_EQ(b[pos], 0x00) << "segment 2 must not carry a maxPointNumber";
// Round-trip: the overflow value goes through the bitmap path.
common::ByteStream dec;
wrap_bytes(b, dec);
float x = 0.0f;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK);
EXPECT_FLOAT_EQ(x, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(dec));
}
// Same overflow layout for double (scaled > INT64_MAX → 1.0e17 * 100).
// The generic walker understands the FLAG + maxPointNumber + segments
// structure; round-trip goes through the bitmap path.
TEST_F(FloatDoubleTS2DIFFCodecTest,
MaxPointNumberOncePerPageDoubleWithOverflow) {
common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
const int row_num = 140;
std::vector<double> data(row_num);
data[0] = 0.5;
data[1] = 1.0e17; // *100 = 1e19 > INT64_MAX → scaled overflow (flag 0)
for (int i = 2; i < row_num; i++) {
data[i] = 0.75 + static_cast<double>(i - 2) * 0.25;
}
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_double_->encode(data[i], out), common::E_OK);
}
ASSERT_EQ(encoder_double_->flush(out), common::E_OK);
std::vector<uint8_t> b = byte_stream_bytes(out);
expect_max_pn_once_per_page(b, true);
common::ByteStream dec;
wrap_bytes(b, dec);
double y = 0.0;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_double_->read_double(y, dec), common::E_OK);
EXPECT_DOUBLE_EQ(y, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_double_->has_remaining(dec));
}
// PageWriter resets the encoder between pages; every page must carry its
// own maxPointNumber prefix (exactly one per page).
TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberPerPageAfterReset) {
const int row_num = 130; // 129 + 1 → two segments per page
std::vector<float> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<float>(i) * 0.25f + 0.5f;
}
common::ByteStream page1(1024, common::MOD_TS2DIFF_OBJ, false);
common::ByteStream page2(1024, common::MOD_TS2DIFF_OBJ, false);
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_float_->encode(data[i], page1), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(page1), common::E_OK);
encoder_float_->reset(); // page boundary
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_float_->encode(data[i], page2), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(page2), common::E_OK);
std::vector<uint8_t> b1 = byte_stream_bytes(page1);
std::vector<uint8_t> b2 = byte_stream_bytes(page2);
ASSERT_FALSE(b1.empty());
ASSERT_FALSE(b2.empty());
EXPECT_EQ(b1[0], 0x02) << "page 1 must start with maxPointNumber=2";
EXPECT_EQ(b2[0], 0x02) << "page 2 must start with maxPointNumber=2";
expect_max_pn_once_per_page(b1, false);
expect_max_pn_once_per_page(b2, false);
// Both pages decode with the same decoder; PageReader calls reset()
// between pages, which must re-arm the per-page prefix state.
common::ByteStream d1;
common::ByteStream d2;
wrap_bytes(b1, d1);
wrap_bytes(b2, d2);
float x = 0.0f;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_float_->read_float(x, d1), common::E_OK);
EXPECT_FLOAT_EQ(x, data[i]) << "page1 row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(d1));
decoder_float_->reset(); // page boundary
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_float_->read_float(x, d2), common::E_OK);
EXPECT_FLOAT_EQ(x, data[i]) << "page2 row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(d2));
}
// Explicit maxPointNumber = 0 (an explicit max_point_number property on
// the Java side): a canonical Form 1 page starting with the 0x00 byte.
// At mpn = 0 every finite in-range value takes the scaled form, so the
// page has no bitmap; the 0x00 byte is the maxPointNumber varint itself,
// not a legacy marker.
TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberZeroPagePrefix) {
const int row_num = 140; // 129 + 11: crosses the block boundary
std::vector<float> data(row_num);
for (int i = 0; i < row_num; i++) {
data[i] = static_cast<float>(i) * 2.0f + 1.0f;
}
common::ByteStream out(1024, common::MOD_TS2DIFF_OBJ, false);
encoder_float_->set_max_point_number(0);
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_float_->encode(data[i], out), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(out), common::E_OK);
std::vector<uint8_t> b = byte_stream_bytes(out);
ASSERT_FALSE(b.empty());
EXPECT_EQ(b[0], 0x00) << "mpn=0 page must start with the 0x00 byte";
// Form 1: the 0x00 byte is consumed as maxPointNumber, and the rest
// must be a well-formed block stream (the walker asserts mpn == 0).
size_t pos = 0;
uint32_t mpn = 999;
ASSERT_TRUE(parse_var_uint(b, pos, mpn));
EXPECT_EQ(mpn, 0u);
common::ByteStream dec;
wrap_bytes(b, dec);
float x = 0.0f;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK);
EXPECT_FLOAT_EQ(x, data[i]) << "row " << i;
}
EXPECT_FALSE(decoder_float_->has_remaining(dec));
// Same for double.
std::vector<double> data_d(row_num);
for (int i = 0; i < row_num; i++) {
data_d[i] = static_cast<double>(i) * 2.0 + 1.0;
}
common::ByteStream out_d(1024, common::MOD_TS2DIFF_OBJ, false);
encoder_double_->set_max_point_number(0);
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(encoder_double_->encode(data_d[i], out_d), common::E_OK);
}
ASSERT_EQ(encoder_double_->flush(out_d), common::E_OK);
std::vector<uint8_t> bd = byte_stream_bytes(out_d);
ASSERT_FALSE(bd.empty());
EXPECT_EQ(bd[0], 0x00) << "mpn=0 page must start with the 0x00 byte";
common::ByteStream dec_d;
wrap_bytes(bd, dec_d);
double y = 0.0;
for (int i = 0; i < row_num; i++) {
ASSERT_EQ(decoder_double_->read_double(y, dec_d), common::E_OK);
EXPECT_DOUBLE_EQ(y, data_d[i]) << "row " << i;
}
EXPECT_FALSE(decoder_double_->has_remaining(dec_d));
}
// The pre-#910 C++ encoder repeated the maxPointNumber at every segment
// boundary. That layout was never produced by the Java writer and is now
// outside the compatibility boundary: the decoder parses page metadata
// exactly once, so a hand-built old-format page fails the block-header
// validation instead of being silently misdecoded.
TEST_F(FloatDoubleTS2DIFFCodecTest, LegacyPerSegmentMaxPNRejected) {
// Build an old-format page by hand: 0x02 prefix before BOTH segments.
const std::vector<float> expected = {0.5f, 0.75f, 1.0f, 1.25f,
1.5f, 1.75f, 2.0f, 2.25f};
common::ByteStream old_fmt(1024, common::MOD_TS2DIFF_OBJ, false);
// Segment 1: 6 values (first 50, five deltas of 25), bit_width 0.
ASSERT_EQ(common::SerializationUtil::write_var_uint(2, old_fmt),
common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(5, old_fmt), common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(0, old_fmt), common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(25, old_fmt), common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(50, old_fmt), common::E_OK);
// Segment 2: 2 values (first 200, one delta of 25) — WITH prefix again.
ASSERT_EQ(common::SerializationUtil::write_var_uint(2, old_fmt),
common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(1, old_fmt), common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(0, old_fmt), common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(25, old_fmt), common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(200, old_fmt),
common::E_OK);
// Segment 1 (a valid single-block Form 1 page) decodes. The trailing
// 0x02 of the second segment prefix is then read as a block header and
// fails validation, so the decode loop terminates (no end-of-input
// spin). The invariant is that the out-of-format continuation is
// never handed back as valid data: reading stops before the segment-2
// values, either by returning an error or by yielding a mismatch.
float x = 0.0f;
size_t good = 0;
for (size_t i = 0; i < expected.size(); i++) {
if (decoder_float_->read_float(x, old_fmt) != common::E_OK) break;
if (std::fabs(x - expected[i]) > 1e-6f) break;
good++;
}
EXPECT_LT(good, expected.size())
<< "out-of-format continuation must not decode to the expected values";
EXPECT_GE(good, 1u) << "segment 1 is a valid page and must still decode";
}
// A truncated page whose block header passes the availability check (the
// check grants slack for the delta_min/first_value fields) but whose
// packed bits never arrive: the scalar read path must terminate instead
// of spinning on a stale buffer.
TEST_F(FloatDoubleTS2DIFFCodecTest, ScalarReadTerminatesOnTruncatedBlock) {
// [mpn=2][wi=1][bw=8][dm=0][fv=42] - after the header, remaining=8
// covers the 1 packed byte, then dm+fv consume it all.
const unsigned char page[] = {
0x02, 0x00, 0x00, 0x00, 0x01, // write_index = 1
0x00, 0x00, 0x00, 0x08, // bit_width = 8
0x00, 0x00, 0x00, 0x00, // delta_min
0x00, 0x00, 0x00, 0x2a, // first_value
};
common::ByteStream dec;
dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
FloatTS2DIFFDecoder decoder;
float x = 0.0f;
ASSERT_EQ(decoder.read_float(x, dec), common::E_OK);
EXPECT_FLOAT_EQ(x, 0.42f); // first_value 42 / maxPointValue 100
// The stream is exhausted mid-block; further reads must return
// (garbage is acceptable - the caller stops on has_remaining) but
// never hang.
for (int i = 0; i < 3; i++) {
decoder.read_float(x, dec);
}
EXPECT_FALSE(decoder.has_remaining(dec));
SUCCEED();
}
TEST_F(FloatDoubleTS2DIFFCodecTest, RejectsBlockBeyondPageValueCount) {
// [overflow marker][count=1][bitmap][mpn=0][wi=INT_MAX][bw=0]...
// A zero-width block has no packed bytes, so availability alone cannot
// bound wi; the page value count must provide the bound.
const unsigned char page[] = {
0xff, 0xff, 0xff, 0xff, 0x07, 0x01, 0x00, 0x00, 0x7f,
0xff, 0xff, 0xff, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01,
};
common::ByteStream dec;
dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
FloatTS2DIFFDecoder decoder;
float value = 0.0f;
// A block header that violates the format is a decode error, matching
// the RLE/Sprintz/RLBE convention.
EXPECT_EQ(decoder.read_float(value, dec), common::E_DECODE_ERR);
}
TEST_F(FloatDoubleTS2DIFFCodecTest, ScalarReadRejectsTruncatedFixedFields) {
// The page metadata is valid, but the block is missing one byte of its
// fixed first-value field.
const unsigned char page[] = {
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
};
common::ByteStream dec;
dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
FloatTS2DIFFDecoder decoder;
float value = 0.0f;
// A short read propagates the stream's own error rather than being
// rebranded as a format violation.
EXPECT_EQ(decoder.read_float(value, dec), common::E_PARTIAL_READ);
}
// A page whose trailing block header is truncated mid-field: the skip
// paths must reject it via the grouped header-read check instead of
// acting on stale stack values.
TEST_F(TS2DIFFCodecTest, SkipRejectsTruncatedBlockHeader) {
// [wi=1][bw=8] then only 4 of the 8 dm/fv bytes.
const unsigned char page[] = {
0x00, 0x00, 0x00, 0x01, // write_index = 1
0x00, 0x00, 0x00, 0x08, // bit_width = 8
0x00, 0x00, 0x00, 0x00, // delta_min
0x00, 0x00, 0x00, // first_value truncated to 3 bytes
};
common::ByteStream dec;
dec.wrap_from(reinterpret_cast<const char*>(page), sizeof(page));
IntTS2DIFFDecoder decoder;
int skipped = 0;
const int rc = decoder.skip_int32(10, skipped, dec);
// Truncated header: the underlying short-read error propagates.
EXPECT_EQ(rc, common::E_PARTIAL_READ);
}
// Java Math.round semantics: floor(x + 0.5), ties towards +infinity.
// -0.125 * 100 = -12.5 exactly (0.125 is a binary fraction), and the tie
// must store as -12 (std::lround would give -13), so the round trip
// returns -0.12 rather than -0.13.
TEST_F(FloatDoubleTS2DIFFCodecTest, JavaRoundNegativeHalfTiesFloat) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const float data[] = {-0.125f, -0.375f, 0.125f, 0.375f};
const float expect[] = {-0.12f, -0.37f, 0.13f, 0.38f};
for (float v : data) {
ASSERT_EQ(encoder_float_->encode(v, out_stream), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK);
float x = 0.0f;
for (int i = 0; i < 4; i++) {
ASSERT_EQ(decoder_float_->read_float(x, out_stream), common::E_OK);
EXPECT_FLOAT_EQ(x, expect[i]) << "value " << i;
}
}
TEST_F(FloatDoubleTS2DIFFCodecTest, JavaRoundNegativeHalfTiesDouble) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
const double data[] = {-2.5, -7.5, 2.5, 7.5}; // mpn=0: scaled == value
const double expect[] = {-2.0, -7.0, 3.0, 8.0};
encoder_double_->set_max_point_number(0);
for (double v : data) {
ASSERT_EQ(encoder_double_->encode(v, out_stream), common::E_OK);
}
ASSERT_EQ(encoder_double_->flush(out_stream), common::E_OK);
double y = 0.0;
for (int i = 0; i < 4; i++) {
ASSERT_EQ(decoder_double_->read_double(y, out_stream), common::E_OK);
EXPECT_DOUBLE_EQ(y, expect[i]) << "value " << i;
}
}
// writeIndex is bounded by availability, not BLOCK_DEFAULT_SIZE: a
// hand-built block larger than 129 values (Java exposes block-size
// constructors) must decode through both the scalar and batch paths.
TEST_F(TS2DIFFCodecTest, LargeBlockBeyondDefaultSizeDecodes) {
const int kCount = 200; // wi = 199 > 128
const int32_t first = 1000;
const int32_t delta = 3;
const int32_t bit_width = 3;
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
ASSERT_EQ(common::SerializationUtil::write_ui32(kCount - 1, out_stream),
common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(bit_width, out_stream),
common::E_OK);
// delta_min: constant deltas of 3 rebase to 0 (raw - min), so the
// decoder reconstructs value += stored(0) + delta_min(3) per step.
ASSERT_EQ(common::SerializationUtil::write_ui32(delta, out_stream),
common::E_OK);
ASSERT_EQ(common::SerializationUtil::write_ui32(first, out_stream),
common::E_OK);
// 199 * 3 bits, MSB-packed, all-zero bytes == rebased delta 0. The
// packed body is ceil(bits / 8) = 75 bytes.
const int packed_bytes = ((kCount - 1) * bit_width + 7) / 8; // 75
for (int i = 0; i < packed_bytes; i++) {
ASSERT_EQ(common::SerializationUtil::write_ui8(0, out_stream),
common::E_OK);
}
// Copy the encoded page into a plain buffer; wrap two read streams
// over it (get_wrapped_buf is only valid for wrap_from streams).
std::vector<uint8_t> page(out_stream.total_size());
uint32_t read_len = 0;
ASSERT_EQ(
out_stream.read_buf(page.data(), out_stream.total_size(), read_len),
common::E_OK);
ASSERT_EQ(read_len, page.size());
// Scalar path.
common::ByteStream dec1;
dec1.wrap_from(reinterpret_cast<const char*>(page.data()),
static_cast<int32_t>(page.size()));
IntTS2DIFFDecoder dec_scalar;
int32_t v = 0;
ASSERT_EQ(dec_scalar.read_int32(v, dec1), common::E_OK);
EXPECT_EQ(v, first);
for (int i = 1; i < kCount; i++) {
ASSERT_EQ(dec_scalar.read_int32(v, dec1), common::E_OK);
EXPECT_EQ(v, first + i * delta) << "row " << i;
}
EXPECT_FALSE(dec_scalar.has_remaining(dec1));
// Batch path.
common::ByteStream dec2;
dec2.wrap_from(reinterpret_cast<const char*>(page.data()),
static_cast<int32_t>(page.size()));
IntTS2DIFFDecoder dec_batch;
std::vector<int32_t> out(kCount);
int actual = 0;
ASSERT_EQ(dec_batch.read_batch_int32(out.data(), kCount, actual, dec2),
common::E_OK);
ASSERT_EQ(actual, kCount);
for (int i = 0; i < kCount; i++) {
EXPECT_EQ(out[i], first + i * delta) << "row " << i;
}
}
// maxPointNumber has no format-level upper bound: a page written with
// mpn = 1000 (Java Math.pow overflows to +inf for the decoder, values
// then divide by inf) stays readable; the decoder must accept the varint
// instead of rejecting it as corrupt.
TEST_F(FloatDoubleTS2DIFFCodecTest, MaxPointNumberAboveLegacyBoundDecodes) {
common::ByteStream out_stream(1024, common::MOD_TS2DIFF_OBJ, false);
encoder_float_->set_max_point_number(1000);
const float data[] = {1.0f, 2.0f, 3.0f};
for (float v : data) {
ASSERT_EQ(encoder_float_->encode(v, out_stream), common::E_OK);
}
ASSERT_EQ(encoder_float_->flush(out_stream), common::E_OK);
// The encoder stores round(v * 10^1000); the double product overflows
// to +inf for every value, so every entry takes the raw-bits form
// (Form 3). The decode restores the exact bits.
std::vector<uint8_t> page(out_stream.total_size());
uint32_t read_len = 0;
ASSERT_EQ(
out_stream.read_buf(page.data(), out_stream.total_size(), read_len),
common::E_OK);
ASSERT_EQ(read_len, page.size());
common::ByteStream dec;
dec.wrap_from(reinterpret_cast<const char*>(page.data()),
static_cast<int32_t>(page.size()));
float x = 0.0f;
for (int i = 0; i < 3; i++) {
ASSERT_EQ(decoder_float_->read_float(x, dec), common::E_OK);
EXPECT_EQ(common::float_to_int(x), common::float_to_int(data[i]))
<< "row " << i;
}
}
} // namespace storage