Skip to content

Commit dca11e3

Browse files
CaideyipiCaideyipi
authored
[To dev/1.3] Fix schema file I/O failure handling and short writes (#19214)
* [To dev/1.3] Fix schema file I/O failure handling and short writes Backport the requested schema-file I/O fix to the 1.3 branch, preserving its failure cleanup, complete writes and regression coverage. (cherry picked from commit e3b6002) * [To dev/1.3] Avoid TODO in schema log recovery comment * Update TsFile snapshot dependency
1 parent f1e915a commit dca11e3

15 files changed

Lines changed: 907 additions & 56 deletions

File tree

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,4 +52,6 @@ private DataNodeSchemaMessages() {}
5252
public static final String GET_TEMPLATE_INFO_ERROR =
5353
"Failed to get schema template [%s] from ConfigNode: %s";
5454
public static final String ILLEGAL_PATH_LOG = "illegal path {}";
55+
public static final String SCHEMA_FILE_WRONG_VERSION =
56+
"SchemaFile with wrong version, please check or upgrade.";
5557
}

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,4 +45,5 @@ private DataNodeSchemaMessages() {}
4545
public static final String GET_ALL_TEMPLATE_ERROR = "从 ConfigNode 获取元数据模板列表失败：%s";
4646
public static final String GET_TEMPLATE_INFO_ERROR = "从 ConfigNode 获取元数据模板 [%s] 失败：%s";
4747
public static final String ILLEGAL_PATH_LOG = "非法路径 {}";
48+
public static final String SCHEMA_FILE_WRONG_VERSION = "SchemaFile 版本错误，请检查或升级。";
4849
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java‎

Lines changed: 92 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.apache.iotdb.consensus.ConsensusFactory;
3030
import org.apache.iotdb.db.conf.IoTDBDescriptor;
3131
import org.apache.iotdb.db.exception.metadata.schemafile.SchemaFileNotExists;
32+
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
3233
import org.apache.iotdb.db.schemaengine.metric.SchemaRegionCachedMetric;
3334
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode;
3435
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.container.ICachedMNodeContainer;
@@ -271,10 +272,42 @@ public Iterator<ICachedMNode> getChildren(ICachedMNode parent)
271272

272273
@Override
273274
public void close() throws IOException {
274-
updateHeaderBuffer();
275-
pageManager.close();
276-
forceChannel();
277-
channel.close();
275+
IOException failure = null;
276+
try {
277+
updateHeaderBuffer();
278+
} catch (IOException e) {
279+
failure = e;
280+
}
281+
try {
282+
pageManager.close();
283+
} catch (IOException e) {
284+
if (failure == null) {
285+
failure = e;
286+
} else {
287+
failure.addSuppressed(e);
288+
}
289+
}
290+
try {
291+
forceChannel();
292+
} catch (IOException e) {
293+
if (failure == null) {
294+
failure = e;
295+
} else {
296+
failure.addSuppressed(e);
297+
}
298+
}
299+
try {
300+
channel.close();
301+
} catch (IOException e) {
302+
if (failure == null) {
303+
failure = e;
304+
} else {
305+
failure.addSuppressed(e);
306+
}
307+
}
308+
if (failure != null) {
309+
throw failure;
310+
}
278311
}
279312

280313
@Override
@@ -353,28 +386,34 @@ private void initFileHeader() throws IOException, MetadataException {
353386
if (channel.size() == 0) {
354387
// new pbtree file
355388
lastPageIndex = 0;
356-
ReadWriteIOUtils.write(lastPageIndex, headerContent);
357-
ReadWriteIOUtils.write(dataTTL, headerContent);
358-
ReadWriteIOUtils.write(isEntity, headerContent);
359-
ReadWriteIOUtils.write(sgNodeTemplateIdWithState, headerContent);
360-
ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent);
361389
lastSGAddr = 0L;
362-
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
363-
} else {
364-
IOUtils.readFully(channel, headerContent);
365-
headerContent.clear();
366-
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
367-
dataTTL = ReadWriteIOUtils.readLong(headerContent);
368-
isEntity = ReadWriteIOUtils.readBool(headerContent);
369-
sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent);
370-
lastSGAddr = ReadWriteIOUtils.readLong(headerContent);
371-
372-
if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) {
373-
channel.close();
374-
throw new MetadataException("SchemaFile with wrong version, please check or upgrade.");
390+
pageManager = null;
391+
try {
392+
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
393+
updateHeaderBuffer();
394+
} catch (IOException | MetadataException e) {
395+
closeAfterInitializationFailure(e);
396+
throw e;
375397
}
398+
} else {
399+
try {
400+
IOUtils.readFully(channel, headerContent);
401+
headerContent.clear();
402+
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
403+
dataTTL = ReadWriteIOUtils.readLong(headerContent);
404+
isEntity = ReadWriteIOUtils.readBool(headerContent);
405+
sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent);
406+
lastSGAddr = ReadWriteIOUtils.readLong(headerContent);
407+
408+
if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) {
409+
throw new MetadataException(DataNodeSchemaMessages.SCHEMA_FILE_WRONG_VERSION);
410+
}
376411

377-
pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
412+
pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
413+
} catch (IOException | MetadataException e) {
414+
closeChannelAfterInitializationFailure(e);
415+
throw e;
416+
}
378417
}
379418
}
380419

@@ -389,7 +428,36 @@ private void updateHeaderBuffer() throws IOException {
389428
ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent);
390429

391430
headerContent.flip();
392-
channel.write(headerContent, 0);
431+
IOUtils.writeFully(channel, headerContent, 0);
432+
}
433+
434+
private void closeAfterInitializationFailure(Exception failure) {
435+
if (pageManager != null) {
436+
try {
437+
pageManager.close();
438+
} catch (IOException closeException) {
439+
failure.addSuppressed(closeException);
440+
}
441+
}
442+
try {
443+
channel.truncate(0);
444+
} catch (IOException cleanupException) {
445+
failure.addSuppressed(cleanupException);
446+
}
447+
try {
448+
channel.force(true);
449+
} catch (IOException cleanupException) {
450+
failure.addSuppressed(cleanupException);
451+
}
452+
closeChannelAfterInitializationFailure(failure);
453+
}
454+
455+
private void closeChannelAfterInitializationFailure(Exception failure) {
456+
try {
457+
channel.close();
458+
} catch (IOException closeException) {
459+
failure.addSuppressed(closeException);
460+
}
393461
}
394462

395463
private void forceChannel() throws IOException {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile;
2020

2121
import org.apache.iotdb.commons.exception.MetadataException;
22+
import org.apache.iotdb.commons.utils.IOUtils;
2223
import org.apache.iotdb.commons.utils.TestOnly;
2324
import org.apache.iotdb.db.exception.metadata.schemafile.SegmentNotFoundException;
2425

@@ -106,7 +107,7 @@ public void syncPageBuffer() {
106107
@Override
107108
public void flushPageToChannel(FileChannel channel) throws IOException {
108109
this.pageBuffer.clear();
109-
channel.write(pageBuffer, SchemaFile.getPageAddress(pageIndex));
110+
IOUtils.writeFully(channel, pageBuffer, SchemaFile.getPageAddress(pageIndex));
110111
dirtyFlag = false;
111112
}
112113

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java‎

Lines changed: 59 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -53,19 +53,28 @@ public class PageIOChannel {
5353
this.channel = channel;
5454
this.pmtFile = pmtFile;
5555
this.readChannel = FileChannel.open(pmtFile.toPath(), StandardOpenOption.READ);
56-
if (flushWithLogging) {
57-
// without RATIS, utilize physical logging for integrity
58-
int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH;
59-
this.logWriter = new SchemaFileLogWriter(logPath);
60-
logCounter = new AtomicInteger(pageAcc);
61-
flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging;
62-
singlePageFlushStrategy = this::flushSinglePageWithLogging;
63-
} else {
64-
// with RATIS enabled, integrity is guaranteed by consensus protocol
65-
logCounter = new AtomicInteger();
66-
logWriter = null;
67-
flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging;
68-
singlePageFlushStrategy = this::flushSinglePageWithoutLogging;
56+
try {
57+
if (flushWithLogging) {
58+
// without RATIS, utilize physical logging for integrity
59+
int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH;
60+
this.logWriter = new SchemaFileLogWriter(logPath);
61+
logCounter = new AtomicInteger(pageAcc);
62+
flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging;
63+
singlePageFlushStrategy = this::flushSinglePageWithLogging;
64+
} else {
65+
// with RATIS enabled, integrity is guaranteed by consensus protocol
66+
logCounter = new AtomicInteger();
67+
logWriter = null;
68+
flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging;
69+
singlePageFlushStrategy = this::flushSinglePageWithoutLogging;
70+
}
71+
} catch (IOException | MetadataException e) {
72+
try {
73+
close();
74+
} catch (IOException closeException) {
75+
e.addSuppressed(closeException);
76+
}
77+
throw e;
6978
}
7079
}
7180

@@ -75,28 +84,57 @@ public void renewLogWriter() throws IOException {
7584
}
7685
}
7786

78-
public void closeLogWriter() throws IOException {
87+
public void close() throws IOException {
88+
IOException failure = null;
7989
if (logWriter != null) {
80-
logWriter.close();
90+
try {
91+
logWriter.close();
92+
} catch (IOException e) {
93+
failure = e;
94+
}
95+
}
96+
try {
97+
readChannel.close();
98+
} catch (IOException e) {
99+
if (failure == null) {
100+
failure = e;
101+
} else {
102+
failure.addSuppressed(e);
103+
}
104+
}
105+
if (failure != null) {
106+
throw failure;
81107
}
82108
}
83109

84110
/** Load bytes from log, deserialize and flush directly into channel, return current length */
85111
private long recoverFromLog(String logPath) throws IOException, MetadataException {
86112
SchemaFileLogReader reader = new SchemaFileLogReader(logPath);
87113
ISchemaPage page;
88-
List<byte[]> res = reader.collectUpdatedEntries();
89-
for (byte[] entry : res) {
90-
// TODO check bytes semantic correctness with CRC32 or other way
91-
page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry));
92-
page.flushPageToChannel(this.channel);
114+
List<byte[]> res;
115+
try {
116+
res = reader.collectUpdatedEntries();
117+
for (byte[] entry : res) {
118+
// Check byte semantics with CRC32 or another method.
119+
page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry));
120+
page.flushPageToChannel(this.channel);
121+
}
122+
} catch (IOException | MetadataException e) {
123+
try {
124+
reader.close();
125+
} catch (IOException closeException) {
126+
e.addSuppressed(closeException);
127+
}
128+
throw e;
93129
}
94130
reader.close();
95131

96132
// complete log file
97133
if (!res.isEmpty()) {
98134
try (FileOutputStream outputStream = new FileOutputStream(logPath, true)) {
99-
outputStream.write(new byte[] {SchemaFileConfig.SF_COMMIT_MARK});
135+
IOUtils.writeFully(
136+
outputStream.getChannel(),
137+
ByteBuffer.wrap(new byte[] {SchemaFileConfig.SF_COMMIT_MARK}));
100138
return outputStream.getChannel().size();
101139
}
102140
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -85,14 +85,23 @@ public abstract class PageManager implements IPageManager {
8585
pageIOChannel = new PageIOChannel(channel, pmtFile, true, logPath);
8686
}
8787

88-
// construct first page if file to init
89-
if (lastPageIndex < 0) {
90-
ISegmentedPage rootPage =
91-
ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0);
92-
rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ);
93-
pagePool.put(rootPage);
94-
rootPage.syncPageBuffer();
95-
rootPage.flushPageToChannel(channel);
88+
try {
89+
// construct first page if file to init
90+
if (lastPageIndex < 0) {
91+
ISegmentedPage rootPage =
92+
ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0);
93+
rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ);
94+
pagePool.put(rootPage);
95+
rootPage.syncPageBuffer();
96+
rootPage.flushPageToChannel(channel);
97+
}
98+
} catch (IOException | MetadataException e) {
99+
try {
100+
pageIOChannel.close();
101+
} catch (IOException closeException) {
102+
e.addSuppressed(closeException);
103+
}
104+
throw e;
96105
}
97106
}
98107

@@ -525,7 +534,7 @@ public void inspect(PrintWriter pw) throws IOException, MetadataException {
525534

526535
@Override
527536
public void close() throws IOException {
528-
pageIOChannel.closeLogWriter();
537+
pageIOChannel.close();
529538
}
530539

531540
// endregion

0 commit comments

Comments
 (0)