Merge pull request #21512 from wangjiaming0909/fix/3.0/TS-3457

fix: page buffer cache flush wrong bytes of data into disk

source/util/src/tpagedbuf.c

@@ -166,7 +166,7 @@ static char* doFlushBufPage(SDiskbasedBuf* pBuf, SPageInfo* pg) {
   char* t = NULL;
   if ((!HAS_DATA_IN_DISK(pg)) || pg->dirty) {
     void* payload = GET_PAYLOAD_DATA(pg);
-    t = doCompressData(payload, pBuf->pageSize, &size, pBuf);
+    t = doCompressData(payload, pBuf->pageSize + sizeof(SFilePage), &size, pBuf);
     if (size < 0) {
       uError("failed to compress data when flushing data to disk, %s", pBuf->id);
       terrno = TSDB_CODE_INVALID_PARA;

source/util/test/CMakeLists.txt

@@ -76,3 +76,11 @@ add_test(
     NAME rbtreeTest
     COMMAND rbtreeTest
 )
+
+# pageBufferTest
+add_executable(pageBufferTest "pageBufferTest.cpp")
+target_link_libraries(pageBufferTest os util gtest_main)
+add_test(
+    NAME pageBufferTest
+    COMMAND pageBufferTest
+)

source/util/test/pageBufferTest.cpp

@@ -157,6 +157,68 @@ void recyclePageTest() {
 
   destroyDiskbasedBuf(pBuf);
 }
+
+int saveDataToPage(SFilePage* pPg, const char* data, uint32_t len) {
+  memcpy(pPg->data + pPg->num, data, len);
+  pPg->num += len;
+  setBufPageDirty(pPg, true);
+  return 0;
+}
+
+bool checkBufVarData(SFilePage* pPg, const char* varData, uint32_t varLen) {
+  const char* start = pPg->data + sizeof(SFilePage);
+  for (uint32_t i = 0; i < (pPg->num - sizeof(SFilePage)) / varLen; ++i) {
+    if (0 != strncmp(start + 6 * i + 3, varData, varLen - 3)) {
+      using namespace std;
+      cout << "pos: " << sizeof(SFilePage) + 6 * i + 3 << " should be " << varData << " but is: " << start + 6 * i + 3
+           << endl;
+      return false;
+    }
+  }
+  return true;
+}
+
+// SPageInfo.pData: |  sizeof(void*) 8 bytes | sizeof(SFilePage) 4 bytes| 4096 bytes |
+//                                           ^
+//                                           |
+//                                       SFilePage: flush to disk from here
+void testFlushAndReadBackBuffer() {
+  SDiskbasedBuf* pBuf = NULL;
+  uint32_t       totalLen = 4096;
+  auto           code = createDiskbasedBuf(&pBuf, totalLen, totalLen * 2, "1", TD_TMP_DIR_PATH);
+  int32_t        pageId = -1;
+  auto*          pPg = (SFilePage*)getNewBufPage(pBuf, &pageId);
+  ASSERT_TRUE(pPg != nullptr);
+  pPg->num = sizeof(SFilePage);
+
+  // save data into page
+  uint32_t len = 6;  // sizeof(SFilePage) + 6 * 682 = 4096
+  // nullbitmap(1) + len(2) + AA\0(3)
+  char* rowData = (char*)taosMemoryCalloc(1, len);
+  *(uint16_t*)(rowData + 2) = (uint16_t)2;
+  rowData[3] = 'A';
+  rowData[4] = 'A';
+
+  while (pPg->num + len <= getBufPageSize(pBuf)) {
+    saveDataToPage(pPg, rowData, len);
+  }
+  ASSERT_EQ(pPg->num, totalLen);
+  ASSERT_TRUE(checkBufVarData(pPg, rowData + 3, len));
+  releaseBufPage(pBuf, pPg);
+
+  // flush to disk
+  int32_t newPgId = -1;
+  pPg = (SFilePage*)getNewBufPage(pBuf, &newPgId);
+  releaseBufPage(pBuf, pPg);
+  pPg = (SFilePage*)getNewBufPage(pBuf, &newPgId);
+  releaseBufPage(pBuf, pPg);
+
+  // reload it from disk
+  pPg = (SFilePage*)getBufPage(pBuf, pageId);
+  ASSERT_TRUE(checkBufVarData(pPg, rowData + 3, len));
+  destroyDiskbasedBuf(pBuf);
+}
+
 }  // namespace
 
 TEST(testCase, resultBufferTest) {
@@ -164,6 +226,7 @@ TEST(testCase, resultBufferTest) {
   simpleTest();
   writeDownTest();
   recyclePageTest();
+  testFlushAndReadBackBuffer();
 }
 
 #pragma GCC diagnostic pop