Skip to content

Commit 1b78fa6

Browse files
Mikael RonströmHery Ramilison
authored andcommitted
BUG#27625172 Step 1
------------------- Conversion to write either 4096 bytes or 8192 bytes in LCP control file wasn't completed. 1) Needed to ensure that reads used partial reads with up to 8192 bytes. Both in Backup block and in TUP block. 2) Needed to check for bytesRead in FSREADREQ of LCP control file to be either 4096 or 8192 bytes. 3) Needed to correct size calculations in convert_ctl_page_to_host and convert_ctl_page_to_network. 4) Needed to fix memcpy to copy actual size instead of always 4096 bytes. Sometimes the memcpy should even copy more than 8192 bytes since it was in expanded format. 5) Removed support of LCP version v1, require initial node restart to upgrade from 7.6.3. 6) Fixed calculations in convert_ctl_page_to_host and convert_ctl_page_to_network to use 3 bytes per part rather than 4 bytes as before. 7) Ensure that word count is calculated using the form (x + 3) / 4 since size is not necessary a multiple of 4 anymore. (cherry picked from commit 86a5d39cd3b1f538e8ffc1f30b22318a864de763)
1 parent 7ffba1d commit 1b78fa6

File tree

8 files changed

+125
-77
lines changed

8 files changed

+125
-77
lines changed

storage/ndb/src/kernel/blocks/backup/Backup.cpp

Lines changed: 101 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -9978,7 +9978,9 @@ Backup::execFSREMOVECONF(Signal* signal)
99789978
* This is recorded using the following variables in the LCP control file.
99799979
*
99809980
* MaxPartPairs:
9981-
* This is the number of parts used in LCPs.
9981+
* This is the maximum number of LCPs that can constitute a recoverable
9982+
* checkpoints. Thus an LCP control file can write at most this many
9983+
* parts. Currently this number is set to 2048.
99829984
*
99839985
* NumPartPairs:
99849986
* This is the number of files used in the restore of this LCP, there is
@@ -10653,7 +10655,7 @@ Backup::execLCP_PREPARE_REQ(Signal* signal)
1065310655
ptr.p->m_first_fragment = true;
1065410656
ptr.p->m_is_lcp_scan_active = false;
1065510657
ptr.p->m_current_lcp_lsn = Uint64(0);
10656-
DEB_LCP(("(%u)TAGS Start new LCP, id: %u", instance(), req.backupId));
10658+
DEB_LCP_STAT(("(%u)TAGS Start new LCP, id: %u", instance(), req.backupId));
1065710659
LocalDeleteLcpFile_list queue(c_deleteLcpFilePool,
1065810660
m_delete_lcp_file_head);
1065910661
ndbrequire(queue.isEmpty())
@@ -10784,7 +10786,7 @@ Backup::lcp_open_ctl_file_done(Signal* signal,
1078410786
Uint32 mem_offset = Uint32((char*)pagePtr.p - (char*)c_startOfPages);
1078510787
req->data.memoryAddress.memoryOffset = mem_offset;
1078610788
req->data.memoryAddress.fileOffset = 0;
10787-
req->data.memoryAddress.size = BackupFormat::NDB_LCP_CTL_FILE_SIZE;
10789+
req->data.memoryAddress.size = BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG;
1078810790

1078910791
sendSignal(NDBFS_REF, GSN_FSREADREQ, signal,
1079010792
FsReadWriteReq::FixedLength + 3, JBA);
@@ -10831,7 +10833,10 @@ Backup::execFSREADCONF(Signal *signal)
1083110833
if (ptr.p->deleteFilePtr == filePtr.i)
1083210834
{
1083310835
jam();
10834-
ndbrequire(filePtr.p->bytesRead == BackupFormat::NDB_LCP_CTL_FILE_SIZE);
10836+
ndbrequire(filePtr.p->bytesRead ==
10837+
BackupFormat::NDB_LCP_CTL_FILE_SIZE_SMALL ||
10838+
filePtr.p->bytesRead ==
10839+
BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG);
1083510840
lcp_read_ctl_file_for_rewrite_done(signal, filePtr);
1083610841
return;
1083710842
}
@@ -11163,7 +11168,8 @@ Backup::lcp_read_ctl_file(Page32Ptr pagePtr,
1116311168
* This information is used to decide which header file to close (the most
1116411169
* recent one) and which header file to use for the next LCP.
1116511170
*/
11166-
ndbrequire(BackupFormat::NDB_LCP_CTL_FILE_SIZE == bytesRead);
11171+
ndbrequire(BackupFormat::NDB_LCP_CTL_FILE_SIZE_SMALL == bytesRead ||
11172+
BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG == bytesRead);
1116711173
if (!convert_ctl_page_to_host(lcpCtlFilePtr))
1116811174
{
1116911175
jam();
@@ -11179,24 +11185,41 @@ Backup::lcp_read_ctl_file(Page32Ptr pagePtr,
1117911185
}
1118011186
}
1118111187

11188+
/**
11189+
* We compress before writing LCP control and after reading it we will
11190+
* decompress the part information. In compressed format we use 3 bytes
11191+
* to store two numbers that can at most be 2048. In uncompressed
11192+
* format each part is a 16-bit unsigned integer.
11193+
*/
11194+
#define BYTES_PER_PART 3
11195+
/**
11196+
* Define the LCP Control file header size, remove the one part pair
11197+
* defined in the common header.
11198+
*/
11199+
#define LCP_CTL_FILE_HEADER_SIZE (sizeof(BackupFormat::LCPCtlFile) - \
11200+
sizeof(BackupFormat::PartPair))
11201+
1118211202
bool
1118311203
Backup::convert_ctl_page_to_host(
1118411204
struct BackupFormat::LCPCtlFile *lcpCtlFilePtr)
1118511205
{
1118611206
Uint32 *pageData = (Uint32*)lcpCtlFilePtr;
1118711207
Uint32 numPartPairs = ntohl(lcpCtlFilePtr->NumPartPairs);
11188-
Uint32 real_bytes_read = (sizeof(*lcpCtlFilePtr) +
11189-
(sizeof(struct BackupFormat::PartPair) *
11190-
(numPartPairs - 1)));
11208+
Uint32 real_bytes_read = LCP_CTL_FILE_HEADER_SIZE +
11209+
(BYTES_PER_PART * numPartPairs);
1119111210

11192-
/* Checksum is calculated on network byte order */
11211+
/* Checksum is calculated on compressed network byte order */
1119311212
if (numPartPairs > BackupFormat::NDB_MAX_LCP_PARTS)
1119411213
{
1119511214
DEB_LCP(("(%u)numPartPairs: %x", instance(), numPartPairs));
1119611215
ndbassert(false);
1119711216
return false;
1119811217
}
11199-
Uint32 words = real_bytes_read / sizeof(Uint32);
11218+
/**
11219+
* Add 3 to ensure that we get also the last word with anything not
11220+
* equal to 0 when changing to word count.
11221+
*/
11222+
Uint32 words = (real_bytes_read + 3) / sizeof(Uint32);
1120011223
Uint32 chksum = 0;
1120111224
for (Uint32 i = 0; i < words; i++)
1120211225
{
@@ -11244,48 +11267,37 @@ Backup::convert_ctl_page_to_host(
1124411267
lcpCtlFilePtr->MaxPartPairs = ntohl(lcpCtlFilePtr->MaxPartPairs);
1124511268
lcpCtlFilePtr->NumPartPairs = ntohl(lcpCtlFilePtr->NumPartPairs);
1124611269

11247-
ndbrequire((2 * BackupFormat::NDB_LCP_CTL_FILE_SIZE) >= real_bytes_read);
11270+
ndbrequire(BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG >= real_bytes_read);
1124811271
ndbrequire(lcpCtlFilePtr->fileHeader.FileType ==
1124911272
BackupFormat::LCP_CTL_FILE);
1125011273
ndbrequire(memcmp(BACKUP_MAGIC, lcpCtlFilePtr->fileHeader.Magic, 8) == 0);
1125111274
ndbrequire(lcpCtlFilePtr->NumPartPairs <= lcpCtlFilePtr->MaxPartPairs);
1125211275
ndbrequire(lcpCtlFilePtr->NumPartPairs > 0);
1125311276
Uint32 total_parts;
11254-
if (lcpCtlFilePtr->fileHeader.BackupVersion == NDBD_USE_PARTIAL_LCP_v2)
11255-
{
11256-
lcpCtlFilePtr->RowCountLow = ntohl(lcpCtlFilePtr->RowCountLow);
11257-
lcpCtlFilePtr->RowCountHigh = ntohl(lcpCtlFilePtr->RowCountHigh);
11258-
total_parts = decompress_part_pairs(lcpCtlFilePtr,
11259-
lcpCtlFilePtr->NumPartPairs,
11260-
&lcpCtlFilePtr->partPairs[0]);
11261-
}
11262-
else
11263-
{
11264-
struct BackupFormat::LCPCtlFile *oldLcpCtlFilePtr =
11265-
(struct BackupFormat::LCPCtlFile*)lcpCtlFilePtr;
11266-
lcpCtlFilePtr->RowCountLow = 0;
11267-
lcpCtlFilePtr->RowCountHigh = 0;
11268-
ndbrequire(lcpCtlFilePtr->fileHeader.BackupVersion ==
11269-
NDBD_USE_PARTIAL_LCP_v1);
11270-
total_parts = decompress_part_pairs(lcpCtlFilePtr,
11271-
lcpCtlFilePtr->NumPartPairs,
11272-
&oldLcpCtlFilePtr->partPairs[0]);
11273-
}
11277+
ndbrequire(lcpCtlFilePtr->fileHeader.BackupVersion >= NDBD_USE_PARTIAL_LCP_v2)
11278+
lcpCtlFilePtr->RowCountLow = ntohl(lcpCtlFilePtr->RowCountLow);
11279+
lcpCtlFilePtr->RowCountHigh = ntohl(lcpCtlFilePtr->RowCountHigh);
11280+
total_parts = decompress_part_pairs(lcpCtlFilePtr,
11281+
lcpCtlFilePtr->NumPartPairs,
11282+
&lcpCtlFilePtr->partPairs[0]);
1127411283
ndbrequire(total_parts <= lcpCtlFilePtr->MaxPartPairs);
1127511284
return true;
1127611285
}
1127711286

1127811287
void
11279-
Backup::convert_ctl_page_to_network(Uint32 *page)
11288+
Backup::convert_ctl_page_to_network(Uint32 *page, Uint32 file_size)
1128011289
{
1128111290
struct BackupFormat::LCPCtlFile *lcpCtlFilePtr =
1128211291
(struct BackupFormat::LCPCtlFile*)page;
1128311292
Uint32 numPartPairs = lcpCtlFilePtr->NumPartPairs;
11284-
Uint32 real_bytes_read = (sizeof(*lcpCtlFilePtr) +
11285-
(sizeof(struct BackupFormat::PartPair) *
11286-
(numPartPairs - 1)));
11293+
Uint32 compressed_bytes_written = LCP_CTL_FILE_HEADER_SIZE +
11294+
(BYTES_PER_PART * numPartPairs);
1128711295

11288-
ndbrequire(BackupFormat::NDB_LCP_CTL_FILE_SIZE >= real_bytes_read);
11296+
/**
11297+
* Add 3 to ensure that we take into account the last word that might
11298+
* filled with only 1 byte of information.
11299+
*/
11300+
ndbrequire(file_size >= (compressed_bytes_written + 3));
1128911301

1129011302
ndbrequire(memcmp(BACKUP_MAGIC, lcpCtlFilePtr->fileHeader.Magic, 8) == 0);
1129111303
ndbrequire(lcpCtlFilePtr->fileHeader.FileType ==
@@ -11334,17 +11346,21 @@ Backup::convert_ctl_page_to_network(Uint32 *page)
1133411346
lcpCtlFilePtr->RowCountLow = htonl(lcpCtlFilePtr->RowCountLow);
1133511347
lcpCtlFilePtr->RowCountHigh = htonl(lcpCtlFilePtr->RowCountHigh);
1133611348

11337-
Uint32 total_parts = compress_part_pairs(lcpCtlFilePtr, numPartPairs);
11349+
Uint32 total_parts = compress_part_pairs(lcpCtlFilePtr,
11350+
numPartPairs,
11351+
file_size);
1133811352
ndbrequire(total_parts <= maxPartPairs);
1133911353

1134011354
/**
11341-
* Checksum is calculated on network byte order.
11355+
* Checksum is calculated on compressed network byte order.
1134211356
* The checksum is calculated without regard to size decreasing due to
1134311357
* compression. This is not a problem since we fill the remainder with
1134411358
* zeroes and XOR doesn't change the checksum with extra zeroes.
11359+
*
11360+
* Add 3 to ensure that we move to word count in a correct manner.
1134511361
*/
1134611362
lcpCtlFilePtr->Checksum = 0;
11347-
Uint32 words = real_bytes_read / sizeof(Uint32);
11363+
Uint32 words = (compressed_bytes_written + 3) / sizeof(Uint32);
1134811364
Uint32 chksum = 0;
1134911365
for (Uint32 i = 0; i < words; i++)
1135011366
{
@@ -11355,7 +11371,8 @@ Backup::convert_ctl_page_to_network(Uint32 *page)
1135511371

1135611372
Uint32
1135711373
Backup::compress_part_pairs(struct BackupFormat::LCPCtlFile *lcpCtlFilePtr,
11358-
Uint32 num_parts)
11374+
Uint32 num_parts,
11375+
Uint32 file_size)
1135911376
{
1136011377
Uint32 total_parts = 0;
1136111378
unsigned char *part_array =
@@ -11390,10 +11407,10 @@ Backup::compress_part_pairs(struct BackupFormat::LCPCtlFile *lcpCtlFilePtr,
1139011407
numParts));
1139111408
}
1139211409
ndbrequire(total_parts == BackupFormat::NDB_MAX_LCP_PARTS);
11393-
unsigned char *start_page = (unsigned char*)lcpCtlFilePtr;
11394-
unsigned char *end_page = start_page + BackupFormat::NDB_LCP_CTL_FILE_SIZE;
11395-
Uint64 remaining_size_64 = end_page - part_array;
11396-
ndbrequire(remaining_size_64 < BackupFormat::NDB_LCP_CTL_FILE_SIZE);
11410+
unsigned char *start_pos = (unsigned char*)lcpCtlFilePtr;
11411+
unsigned char *end_pos = start_pos + file_size;
11412+
Uint64 remaining_size_64 = end_pos - part_array;
11413+
ndbrequire(remaining_size_64 < file_size);
1139711414
Uint32 remaining_size = Uint32(remaining_size_64);
1139811415
memset(part_array, 0, remaining_size);
1139911416
return total_parts;
@@ -11646,9 +11663,38 @@ Backup::lcp_copy_ctl_page(BackupRecordPtr ptr)
1164611663
c_backupFilePool.getPtr(recent_file_ptr, ptr.p->prepareCtlFilePtr[recent]);
1164711664
file_ptr.p->pages.getPtr(page_ptr, 0);
1164811665
recent_file_ptr.p->pages.getPtr(recent_page_ptr, 0);
11649-
memcpy(page_ptr.p,
11650-
recent_page_ptr.p,
11651-
BackupFormat::NDB_LCP_CTL_FILE_SIZE);
11666+
/**
11667+
* Important to consider here that the page is currently in expanded
11668+
* format. So before we copy it we calculate how much to copy.
11669+
*/
11670+
{
11671+
struct BackupFormat::LCPCtlFile *lcpCtlFilePtr =
11672+
(struct BackupFormat::LCPCtlFile*)recent_page_ptr.p;
11673+
Uint32 num_parts = lcpCtlFilePtr->NumPartPairs;
11674+
Uint32 size_to_copy = LCP_CTL_FILE_HEADER_SIZE;
11675+
size_to_copy += (num_parts * sizeof(struct BackupFormat::PartPair));
11676+
memcpy(page_ptr.p,
11677+
recent_page_ptr.p,
11678+
size_to_copy);
11679+
}
11680+
#ifdef VM_TRACE
11681+
{
11682+
struct BackupFormat::LCPCtlFile *lcpCtlFilePtr =
11683+
(struct BackupFormat::LCPCtlFile*)page_ptr.p;
11684+
jam();
11685+
Uint32 total_parts = 0;
11686+
Uint32 num_parts = lcpCtlFilePtr->NumPartPairs;
11687+
jamLine(num_parts);
11688+
for (Uint32 i = 0; i < num_parts; i++)
11689+
{
11690+
Uint32 parts = lcpCtlFilePtr->partPairs[i].numParts;
11691+
total_parts += parts;
11692+
jamLine(parts);
11693+
}
11694+
jam();
11695+
ndbassert(total_parts == BackupFormat::NDB_MAX_LCP_PARTS);
11696+
}
11697+
#endif
1165211698
}
1165311699

1165411700
void
@@ -13059,21 +13105,20 @@ Backup::lcp_write_ctl_file_to_disk(Signal *signal,
1305913105
struct BackupFormat::LCPCtlFile *lcpCtlFilePtr =
1306013106
(struct BackupFormat::LCPCtlFile*)pagePtr.p;
1306113107
Uint32 num_parts = lcpCtlFilePtr->NumPartPairs;
13062-
Uint32 size_written = (sizeof(BackupFormat::LCPCtlFile) -
13063-
sizeof(BackupFormat::PartPair)) +
13064-
(3 * num_parts + 3);
13065-
if (size_written > BackupFormat::NDB_LCP_CTL_FILE_SIZE)
13108+
Uint32 file_size = LCP_CTL_FILE_HEADER_SIZE +
13109+
(3 * num_parts + 3);
13110+
if (file_size > BackupFormat::NDB_LCP_CTL_FILE_SIZE_SMALL)
1306613111
{
1306713112
jam();
1306813113
DEB_LCP(("(%u)Writing 8192 byte control file", instance()));
13069-
size_written = BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG;
13114+
file_size = BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG;
1307013115
}
1307113116
else
1307213117
{
1307313118
jam();
13074-
size_written = BackupFormat::NDB_LCP_CTL_FILE_SIZE;
13119+
file_size = BackupFormat::NDB_LCP_CTL_FILE_SIZE_SMALL;
1307513120
}
13076-
convert_ctl_page_to_network((Uint32*)pagePtr.p);
13121+
convert_ctl_page_to_network((Uint32*)pagePtr.p, file_size);
1307713122
filePtr.p->m_flags |= BackupFile::BF_WRITING;
1307813123
FsReadWriteReq* req = (FsReadWriteReq*)signal->getDataPtrSend();
1307913124
req->userPointer = filePtr.i;
@@ -13089,7 +13134,7 @@ Backup::lcp_write_ctl_file_to_disk(Signal *signal,
1308913134
Uint32 mem_offset = Uint32((char*)pagePtr.p - (char*)c_startOfPages);
1309013135
req->data.memoryAddress.memoryOffset = mem_offset;
1309113136
req->data.memoryAddress.fileOffset = 0;
13092-
req->data.memoryAddress.size = size_written;
13137+
req->data.memoryAddress.size = file_size;
1309313138

1309413139
sendSignal(NDBFS_REF, GSN_FSWRITEREQ, signal,
1309513140
FsReadWriteReq::FixedLength + 3, JBA);
@@ -13591,7 +13636,7 @@ Backup::lcp_read_ctl_file_for_rewrite(Signal *signal,
1359113636
Uint32 mem_offset = Uint32(((char*)pagePtr.p) - ((char*)c_startOfPages));
1359213637
req->data.memoryAddress.memoryOffset = mem_offset;
1359313638
req->data.memoryAddress.fileOffset = 0;
13594-
req->data.memoryAddress.size = BackupFormat::NDB_LCP_CTL_FILE_SIZE;
13639+
req->data.memoryAddress.size = BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG;
1359513640

1359613641
sendSignal(NDBFS_REF, GSN_FSREADREQ, signal,
1359713642
FsReadWriteReq::FixedLength + 3, JBA);

storage/ndb/src/kernel/blocks/backup/Backup.hpp

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1154,12 +1154,14 @@ class Backup : public SimulatedBlock
11541154
void lcp_write_ctl_file(Signal*, BackupRecordPtr);
11551155
void lcp_write_ctl_file_to_disk(Signal*, BackupFilePtr, Page32Ptr);
11561156
void lcp_init_ctl_file(Page32Ptr pagePtr);
1157-
Uint32 compress_part_pairs(struct BackupFormat::LCPCtlFile*, Uint32);
1157+
Uint32 compress_part_pairs(struct BackupFormat::LCPCtlFile*,
1158+
Uint32 numPartPairs,
1159+
Uint32 file_size);
11581160
Uint32 decompress_part_pairs(struct BackupFormat::LCPCtlFile*,
11591161
Uint32,
11601162
struct BackupFormat::PartPair*);
11611163
bool convert_ctl_page_to_host(struct BackupFormat::LCPCtlFile*);
1162-
void convert_ctl_page_to_network(Uint32*);
1164+
void convert_ctl_page_to_network(Uint32*, Uint32 file_size);
11631165
void handle_idle_lcp(Signal*, BackupRecordPtr);
11641166
Uint32 calculate_min_parts(Uint64 row_count,
11651167
Uint64 row_change_count,

storage/ndb/src/kernel/blocks/backup/BackupFormat.hpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
Copyright (c) 2003, 2017, Oracle and/or its affiliates. All rights reserved.
2+
Copyright (c) 2003, 2018, Oracle and/or its affiliates. All rights reserved.
33
44
This program is free software; you can redistribute it and/or modify
55
it under the terms of the GNU General Public License as published by
@@ -32,7 +32,7 @@ struct BackupFormat {
3232
static const Uint32 NDB_MAX_LCP_PARTS_PER_ROUND =
3333
NDB_MAX_LCP_PARTS / NDB_MAX_FILES_PER_LCP;
3434
static const Uint32 NDB_MAX_LCP_FILES = 2064;
35-
static const Uint32 NDB_LCP_CTL_FILE_SIZE = 4096;
35+
static const Uint32 NDB_LCP_CTL_FILE_SIZE_SMALL = 4096;
3636
static const Uint32 NDB_LCP_CTL_FILE_SIZE_BIG = 8192;
3737

3838
enum RecordType

storage/ndb/src/kernel/blocks/dbtup/Dbtup.hpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1164,7 +1164,7 @@ TupTriggerData_pool c_triggerPool;
11641164
State tableStatus;
11651165
Local_key m_default_value_location;
11661166
};
1167-
Uint32 m_read_ctl_file_data[BackupFormat::NDB_LCP_CTL_FILE_SIZE / 4];
1167+
Uint32 m_read_ctl_file_data[BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG / 4];
11681168
/*
11691169
It is more space efficient to store dynamic fixed-size attributes
11701170
of more than about 16 words as variable-sized internally.

storage/ndb/src/kernel/blocks/dbtup/DbtupGen.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
Copyright (c) 2003, 2017, Oracle and/or its affiliates. All rights reserved.
2+
Copyright (c) 2003, 2018, Oracle and/or its affiliates. All rights reserved.
33
44
This program is free software; you can redistribute it and/or modify
55
it under the terms of the GNU General Public License as published by
@@ -522,7 +522,7 @@ void Dbtup::execREAD_CONFIG_REQ(Signal* signal)
522522
*/
523523
NewVARIABLE *bat = allocateBat(1);
524524
bat[0].WA = &m_read_ctl_file_data[0];
525-
bat[0].nrr = BackupFormat::NDB_LCP_CTL_FILE_SIZE;
525+
bat[0].nrr = BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG;
526526
}
527527

528528
void Dbtup::initRecords()

storage/ndb/src/kernel/blocks/dbtup/DbtupMeta.cpp

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2615,7 +2615,7 @@ Dbtup::lcp_read_ctl_file(Signal *signal,
26152615
FsReadWriteReq::setPartialReadFlag(req->operationFlag, 1);
26162616
req->data.memoryAddress.memoryOffset = 0;
26172617
req->data.memoryAddress.fileOffset = 0;
2618-
req->data.memoryAddress.size = BackupFormat::NDB_LCP_CTL_FILE_SIZE;
2618+
req->data.memoryAddress.size = BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG;
26192619
sendSignal(NDBFS_REF, GSN_FSREADREQ, signal,
26202620
FsReadWriteReq::FixedLength + 3, JBA);
26212621
}
@@ -2716,8 +2716,10 @@ Dbtup::handle_ctl_info(TablerecPtr tabPtr,
27162716
{
27172717
BackupFormat::LCPCtlFile *lcpCtlFilePtr =
27182718
(BackupFormat::LCPCtlFile*)&m_read_ctl_file_data[0];
2719-
ndbassert(bytesRead == BackupFormat::NDB_LCP_CTL_FILE_SIZE);
2720-
if (bytesRead != BackupFormat::NDB_LCP_CTL_FILE_SIZE ||
2719+
ndbassert(bytesRead == BackupFormat::NDB_LCP_CTL_FILE_SIZE_SMALL ||
2720+
bytesRead == BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG);
2721+
if ((bytesRead != BackupFormat::NDB_LCP_CTL_FILE_SIZE_SMALL &&
2722+
bytesRead != BackupFormat::NDB_LCP_CTL_FILE_SIZE_BIG) ||
27212723
!c_backup->convert_ctl_page_to_host(lcpCtlFilePtr))
27222724
{
27232725
jam();

0 commit comments

Comments
 (0)