fix(cpp): correct aligned all-null pages and page advancement

cpp/src/file/restorable_tsfile_io_writer.cc

@@ -54,6 +54,7 @@ namespace {
 const int HEADER_LEN = MAGIC_STRING_TSFILE_LEN + 1;  // magic + version
 const int BUF_SIZE = 4096;
 const unsigned char kTimeChunkTypeMask = 0x80;
+const unsigned char kValueChunkTypeMask = 0x40;
 
 // -----------------------------------------------------------------------------
 // Self-check helpers: read file, parse chunk header, recover chunk statistics
@@ -263,6 +264,13 @@ static int recover_chunk_statistic(
                                  kTimeChunkTypeMask) != 0;
     PageHeader ph;
     int ret = ph.deserialize_from(bs, false, chdr.data_type_);
+    if (ret == common::E_OK && ph.uncompressed_size_ == 0 &&
+        (static_cast<unsigned char>(chdr.chunk_type_) & kValueChunkTypeMask) !=
+            0) {
+        // A single all-null aligned value page has no payload or statistics.
+        return bs.remaining_size() == 0 ? common::E_OK
+                                        : common::E_TSFILE_CORRUPTED;
+    }
     if (ret != common::E_OK || ph.compressed_size_ == 0 ||
         bs.remaining_size() < ph.compressed_size_) {
         // Align with Java selfCheck behavior: malformed/incomplete page in this

cpp/src/reader/aligned_chunk_reader.cc

@@ -699,11 +699,15 @@ int AlignedChunkReader::decode_time_value_buf_into_tsblock(
             value_compressor_->after_uncompress(value_uncompressed_buf_);
             value_uncompressed_buf_ = nullptr;
         }
-        if (!prev_value_page_not_finish()) {
-            value_in_.reset();
-        }
         if (!prev_time_page_not_finish()) {
+            // The time page determines the aligned value page's end. Even
+            // after decoding every value, Gorilla can leave a padding byte
+            // unread, which must not prevent advancing to the next page.
+            value_in_.reset();
+            value_decoder_->reset();
             time_in_.reset();
+        } else if (!prev_value_page_not_finish()) {
+            value_in_.reset();
         }
         value_page_col_notnull_bitmap_.clear();
         value_page_col_notnull_bitmap_.shrink_to_fit();

cpp/src/writer/value_chunk_writer.cc

@@ -159,6 +159,10 @@ void ValueChunkWriter::save_first_page_data(
 int ValueChunkWriter::write_first_page_data(ByteStream& pages_data,
                                             bool with_statistic) {
     int ret = E_OK;
+    if (first_page_data_.uncompressed_size_ == 0) {
+        // The empty first page is already complete in chunk_data_.
+        return ret;
+    }
     if (with_statistic &&
         RET_FAIL(first_page_statistic_->serialize_to(pages_data))) {
     } else if (RET_FAIL(

cpp/src/writer/value_page_writer.cc

@@ -151,11 +151,16 @@ int ValuePageWriter::write_to_chunk(ByteStream& pages_data, bool write_header,
               << pages_data.total_size() << " of chunk_data." << std::endl;
 #endif
     int ret = E_OK;
-    if (RET_FAIL(prepare_end_page())) {
-        return ret;
-    }
-    if (RET_FAIL(cur_page_data_.init(col_notnull_bitmap_out_stream_,
-                                     value_out_stream_, compressor_, size_))) {
+    if (statistic_->count_ == 0) {
+        // Match the aligned empty-page format: a single zero varint, with
+        // no compressed size, statistics, bitmap, or codec payload.
+        cur_page_data_ = ValuePageData();
+    } else {
+        if (RET_FAIL(prepare_end_page())) {
+            return ret;
+        }
+        ret = cur_page_data_.init(col_notnull_bitmap_out_stream_,
+                                  value_out_stream_, compressor_, size_);
     }
     col_notnull_bitmap_.clear();
     size_ = 0;
@@ -164,21 +169,24 @@ int ValuePageWriter::write_to_chunk(ByteStream& pages_data, bool write_header,
     if (IS_SUCC(ret) && write_header) {
         if (RET_FAIL(SerializationUtil::write_var_uint(
                 cur_page_data_.uncompressed_size_, pages_data))) {
-        } else if (RET_FAIL(SerializationUtil::write_var_uint(
+        } else if (cur_page_data_.uncompressed_size_ != 0 &&
+                   RET_FAIL(SerializationUtil::write_var_uint(
                        cur_page_data_.compressed_size_, pages_data))) {
         }
     }
     // std::cout << "ValuePageWriter::write_to_chunk after write_header. pos="
     // << pages_data.total_size() << std::endl;
-    if (IS_SUCC(ret) && write_statistic) {
+    if (IS_SUCC(ret) && write_statistic &&
+        cur_page_data_.uncompressed_size_ != 0) {
         if (RET_FAIL(statistic_->serialize_to(pages_data))) {
         }
     }
     // std::cout << "ValuePageWriter::write_to_chunk after write_stat. pos=" <<
     // pages_data.total_size() << std::endl; DEBUG_print_byte_stream("In
     // ValuePageWriter::write_to_chunk, before writer page data: pages_data = ",
     // pages_data);
-    if (IS_SUCC(ret) && write_data_to_chunk_data) {
+    if (IS_SUCC(ret) && write_data_to_chunk_data &&
+        cur_page_data_.compressed_size_ != 0) {
         // DEBUG_hex_dump_buf("cur_page_data_.compressed_buf_ = ",
         // cur_page_data_.compressed_buf_, cur_page_data_.compressed_size_);
         if (RET_FAIL(pages_data.write_buf(cur_page_data_.compressed_buf_,

cpp/test/reader/aligned_null_page_test.cc

@@ -0,0 +1,235 @@
+/*
+ * 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 <gtest/gtest.h>
+
+#include <algorithm>
+#include <cstdio>
+#include <memory>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include "common/path.h"
+#include "reader/qds_without_timegenerator.h"
+#include "reader/tsfile_reader.h"
+#include "writer/tsfile_writer.h"
+
+namespace storage {
+namespace {
+
+using namespace common;
+
+// Leading and intermediate all-null value pages must not prevent reading
+// subsequent pages, regardless of the value codec or write API.
+class AlignedNullPageReadTest
+    : public ::testing::TestWithParam<std::tuple<TSDataType, uint32_t, int>> {
+   protected:
+    void SetUp() override {
+        ASSERT_EQ(libtsfile_init(), E_OK);
+        saved_page_rows_ = g_config_value_.page_writer_max_point_num_;
+        saved_block_memory_ = g_config_value_.tsblock_max_memory_;
+        g_config_value_.page_writer_max_point_num_ = std::get<1>(GetParam());
+        // The larger pages span multiple result blocks.
+        g_config_value_.tsblock_max_memory_ = 4096;
+        file_ = "aligned_null_page_" +
+                std::to_string(static_cast<int>(std::get<0>(GetParam()))) +
+                "_" + std::to_string(std::get<1>(GetParam())) + "_" +
+                std::to_string(std::get<2>(GetParam())) + ".tsfile";
+        std::remove(file_.c_str());
+    }
+
+    void TearDown() override {
+        std::remove(file_.c_str());
+        g_config_value_.page_writer_max_point_num_ = saved_page_rows_;
+        g_config_value_.tsblock_max_memory_ = saved_block_memory_;
+        libtsfile_destroy();
+    }
+
+    bool is_null(int row) const {
+        int page = row / std::get<1>(GetParam());
+        return page == 0 || page == 2 || page == 3 || page == 6;
+    }
+
+    void add_value(TsRecord& record, int row) {
+        switch (std::get<0>(GetParam())) {
+            case INT32:
+                record.add_point("value", static_cast<int32_t>(1000 + row));
+                break;
+            case INT64:
+                record.add_point("value", static_cast<int64_t>(1000 + row));
+                break;
+            case FLOAT:
+                record.add_point("value", static_cast<float>(1000.25 + row));
+                break;
+            case DOUBLE:
+                record.add_point("value", 1000.25 + row);
+                break;
+            case STRING:
+                record_string_ = "value-" + std::to_string(row);
+                record.add_point("value", String(record_string_));
+                break;
+            default:
+                FAIL() << "Unexpected type";
+        }
+    }
+
+    void add_value(Tablet& tablet, int index, int row) {
+        switch (std::get<0>(GetParam())) {
+            case INT32:
+                ASSERT_EQ(tablet.add_value(index, 1,
+                                           static_cast<int32_t>(1000 + row)),
+                          E_OK);
+                break;
+            case INT64:
+                ASSERT_EQ(tablet.add_value(index, 1,
+                                           static_cast<int64_t>(1000 + row)),
+                          E_OK);
+                break;
+            case FLOAT:
+                ASSERT_EQ(tablet.add_value(index, 1,
+                                           static_cast<float>(1000.25 + row)),
+                          E_OK);
+                break;
+            case DOUBLE:
+                ASSERT_EQ(tablet.add_value(index, 1, 1000.25 + row), E_OK);
+                break;
+            case STRING: {
+                std::string value = "value-" + std::to_string(row);
+                ASSERT_EQ(tablet.add_value(index, 1, value.c_str()), E_OK);
+                break;
+            }
+            default:
+                FAIL() << "Unexpected type";
+        }
+    }
+
+    void check_value(Field* field, int row) {
+        if (is_null(row)) {
+            EXPECT_EQ(field->type_, NULL_TYPE) << "row=" << row;
+            return;
+        }
+        ASSERT_EQ(field->type_, std::get<0>(GetParam())) << "row=" << row;
+        switch (field->type_) {
+            case INT32:
+                EXPECT_EQ(field->value_.ival_, 1000 + row);
+                break;
+            case INT64:
+                EXPECT_EQ(field->value_.lval_, 1000 + row);
+                break;
+            case FLOAT:
+                EXPECT_FLOAT_EQ(field->value_.fval_, 1000.25f + row);
+                break;
+            case DOUBLE:
+                EXPECT_DOUBLE_EQ(field->value_.dval_, 1000.25 + row);
+                break;
+            case STRING:
+                EXPECT_EQ(field->get_string_value()->to_std_string(),
+                          "value-" + std::to_string(row));
+                break;
+            default:
+                FAIL() << "Unexpected type";
+        }
+    }
+
+    std::string file_;
+    std::string record_string_;
+    uint32_t saved_page_rows_ = 0;
+    uint32_t saved_block_memory_ = 0;
+};
+
+TEST_P(AlignedNullPageReadTest,
+       ReadsValuesAfterLeadingAndIntermediateNullPages) {
+    const TSDataType type = std::get<0>(GetParam());
+    const TSEncoding encoding = type == STRING ? DICTIONARY : GORILLA;
+    const std::string device = "root.null_pages";
+    const int rows = 7 * std::get<1>(GetParam());
+    const int mode = std::get<2>(GetParam());
+    {
+        TsFileWriter writer;
+        ASSERT_EQ(writer.open(file_), E_OK);
+        std::vector<MeasurementSchema*> schemas{
+            new MeasurementSchema("anchor", INT64, PLAIN, UNCOMPRESSED),
+            new MeasurementSchema("value", type, encoding, UNCOMPRESSED)};
+        ASSERT_EQ(writer.register_aligned_timeseries(device, schemas), E_OK);
+        for (int row = 0; row < rows;) {
+            if (mode == 0 || (mode == 2 && row % 3 == 0)) {
+                TsRecord record(row, device);
+                record.add_point("anchor", static_cast<int64_t>(row));
+                if (is_null(row))
+                    record.points_.emplace_back("value");
+                else
+                    add_value(record, row);
+                ASSERT_EQ(writer.write_record_aligned(record), E_OK);
+                ++row;
+            } else {
+                const int count = std::min(11, rows - row);
+                auto schema =
+                    std::make_shared<std::vector<MeasurementSchema>>();
+                schema->emplace_back("anchor", INT64, PLAIN, UNCOMPRESSED);
+                schema->emplace_back("value", type, encoding, UNCOMPRESSED);
+                Tablet tablet(device, schema, count);
+                ASSERT_EQ(tablet.err_code_, E_OK);
+                for (int index = 0; index < count; ++index) {
+                    ASSERT_EQ(tablet.add_timestamp(index, row + index), E_OK);
+                    ASSERT_EQ(tablet.add_value(
+                                  index, 0, static_cast<int64_t>(row + index)),
+                              E_OK);
+                    if (!is_null(row + index))
+                        add_value(tablet, index, row + index);
+                }
+                ASSERT_EQ(writer.write_tablet_aligned(tablet), E_OK);
+                row += count;
+            }
+        }
+        ASSERT_EQ(writer.flush(), E_OK);
+        ASSERT_EQ(writer.close(), E_OK);
+    }
+
+    TsFileReader reader;
+    ASSERT_EQ(reader.open(file_), E_OK);
+    ResultSet* result = nullptr;
+    std::vector<Path> paths{Path(device + ".anchor"), Path(device + ".value")};
+    ASSERT_EQ(reader.query(QueryExpression::create(paths, nullptr), result),
+              E_OK);
+    int row = 0;
+    bool more = false;
+    while (true) {
+        ASSERT_EQ(result->next(more), E_OK);
+        if (!more) break;
+        auto* record = result->get_row_record();
+        ASSERT_LT(row, rows);
+        EXPECT_EQ(record->get_timestamp(), row);
+        EXPECT_EQ(record->get_field(1)->value_.lval_, row);
+        check_value(record->get_field(2), row);
+        ++row;
+    }
+    EXPECT_EQ(row, rows);
+    reader.destroy_query_data_set(result);
+    ASSERT_EQ(reader.close(), E_OK);
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    CodecsAndPageBoundaries, AlignedNullPageReadTest,
+    ::testing::Combine(::testing::Values(INT32, INT64, FLOAT, DOUBLE, STRING),
+                       ::testing::Values(1u, 7u, 256u),
+                       ::testing::Values(0, 1, 2)));
+
+}  // namespace
+}  // namespace storage

cpp/test/tools/golden/empty_field_meta_csv.txt

@@ -1,2 +1,2 @@
 size_bytes,format_version,model
-457,4,table
+451,4,table

cpp/test/tools/golden/empty_field_meta_ndjson.txt

@@ -1 +1 @@
-{"size_bytes":"457","format_version":"4","model":"table"}
+{"size_bytes":"451","format_version":"4","model":"table"}

cpp/test/tools/golden/empty_field_meta_table.txt

@@ -1,2 +1,2 @@
 size_bytes  format_version  model
-457         4               table
+451         4               table