Skip to content

[To dev/1.3] Fix schema file I/O failure handling and short writes #19214

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Caideyipi merged 3 commits into from
Nov 11, 2026
Merged

[To dev/1.3] Fix schema file I/O failure handling and short writes #19214

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
96cd1a3
[To dev/1.3] Fix schema file I/O failure handling and short writes
Caideyipi Nov 11, 2026
0ae5cc2
[To dev/1.3] Avoid TODO in schema log recovery comment
Caideyipi Nov 11, 2026
1ce0f0e
Update TsFile snapshot dependency
Caideyipi Nov 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,4 +52,6 @@ private DataNodeSchemaMessages() {}
public static final String GET_TEMPLATE_INFO_ERROR =
"Failed to get schema template [%s] from ConfigNode: %s";
public static final String ILLEGAL_PATH_LOG = "illegal path {}";
public static final String SCHEMA_FILE_WRONG_VERSION =
"SchemaFile with wrong version, please check or upgrade.";
}
1 change: 1 addition & 0 deletions iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeSchemaMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,4 +45,5 @@ private DataNodeSchemaMessages() {}
public static final String GET_ALL_TEMPLATE_ERROR = "从 ConfigNode 获取元数据模板列表失败：%s";
public static final String GET_TEMPLATE_INFO_ERROR = "从 ConfigNode 获取元数据模板 [%s] 失败：%s";
public static final String ILLEGAL_PATH_LOG = "非法路径 {}";
public static final String SCHEMA_FILE_WRONG_VERSION = "SchemaFile 版本错误，请检查或升级。";
}
116 changes: 92 additions & 24 deletions ...rg/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.metadata.schemafile.SchemaFileNotExists;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
import org.apache.iotdb.db.schemaengine.metric.SchemaRegionCachedMetric;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.ICachedMNode;
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.mnode.container.ICachedMNodeContainer;
Expand Down Expand Up @@ -271,10 +272,42 @@ public Iterator<ICachedMNode> getChildren(ICachedMNode parent)

@Override
public void close() throws IOException {
updateHeaderBuffer();
pageManager.close();
forceChannel();
channel.close();
IOException failure = null;
try {
updateHeaderBuffer();
} catch (IOException e) {
failure = e;
}
try {
pageManager.close();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
try {
forceChannel();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
try {
channel.close();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
if (failure != null) {
throw failure;
}
}

@Override
Expand Down Expand Up @@ -353,28 +386,34 @@ private void initFileHeader() throws IOException, MetadataException {
if (channel.size() == 0) {
// new pbtree file
lastPageIndex = 0;
ReadWriteIOUtils.write(lastPageIndex, headerContent);
ReadWriteIOUtils.write(dataTTL, headerContent);
ReadWriteIOUtils.write(isEntity, headerContent);
ReadWriteIOUtils.write(sgNodeTemplateIdWithState, headerContent);
ReadWriteIOUtils.write(SchemaFileConfig.SCHEMA_FILE_VERSION, headerContent);
lastSGAddr = 0L;
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
} else {
IOUtils.readFully(channel, headerContent);
headerContent.clear();
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
dataTTL = ReadWriteIOUtils.readLong(headerContent);
isEntity = ReadWriteIOUtils.readBool(headerContent);
sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent);
lastSGAddr = ReadWriteIOUtils.readLong(headerContent);

if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) {
channel.close();
throw new MetadataException("SchemaFile with wrong version, please check or upgrade.");
pageManager = null;
try {
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
updateHeaderBuffer();
} catch (IOException | MetadataException e) {
closeAfterInitializationFailure(e);
throw e;
}
} else {
try {
IOUtils.readFully(channel, headerContent);
headerContent.clear();
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
dataTTL = ReadWriteIOUtils.readLong(headerContent);
isEntity = ReadWriteIOUtils.readBool(headerContent);
sgNodeTemplateIdWithState = ReadWriteIOUtils.readInt(headerContent);
lastSGAddr = ReadWriteIOUtils.readLong(headerContent);

if (ReadWriteIOUtils.readInt(headerContent) != SchemaFileConfig.SCHEMA_FILE_VERSION) {
throw new MetadataException(DataNodeSchemaMessages.SCHEMA_FILE_WRONG_VERSION);
}

pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
pageManager = new BTreePageManager(channel, pmtFile, lastPageIndex, logPath);
} catch (IOException | MetadataException e) {
closeChannelAfterInitializationFailure(e);
throw e;
}
}
}

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

headerContent.flip();
channel.write(headerContent, 0);
IOUtils.writeFully(channel, headerContent, 0);
}

private void closeAfterInitializationFailure(Exception failure) {
if (pageManager != null) {
try {
pageManager.close();
} catch (IOException closeException) {
failure.addSuppressed(closeException);
}
}
try {
channel.truncate(0);
} catch (IOException cleanupException) {
failure.addSuppressed(cleanupException);
}
try {
channel.force(true);
} catch (IOException cleanupException) {
failure.addSuppressed(cleanupException);
}
closeChannelAfterInitializationFailure(failure);
}

private void closeChannelAfterInitializationFailure(Exception failure) {
try {
channel.close();
} catch (IOException closeException) {
failure.addSuppressed(closeException);
}
}

private void forceChannel() throws IOException {
Expand Down
3 changes: 2 additions & 1 deletion ...rg/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaPage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile;

import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.utils.IOUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.exception.metadata.schemafile.SegmentNotFoundException;

Expand Down Expand Up @@ -106,7 +107,7 @@ public void syncPageBuffer() {
@Override
public void flushPageToChannel(FileChannel channel) throws IOException {
this.pageBuffer.clear();
channel.write(pageBuffer, SchemaFile.getPageAddress(pageIndex));
IOUtils.writeFully(channel, pageBuffer, SchemaFile.getPageAddress(pageIndex));
dirtyFlag = false;
}

Expand Down
80 changes: 59 additions & 21 deletions ...otdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,19 +53,28 @@ public class PageIOChannel {
this.channel = channel;
this.pmtFile = pmtFile;
this.readChannel = FileChannel.open(pmtFile.toPath(), StandardOpenOption.READ);
if (flushWithLogging) {
// without RATIS, utilize physical logging for integrity
int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH;
this.logWriter = new SchemaFileLogWriter(logPath);
logCounter = new AtomicInteger(pageAcc);
flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging;
singlePageFlushStrategy = this::flushSinglePageWithLogging;
} else {
// with RATIS enabled, integrity is guaranteed by consensus protocol
logCounter = new AtomicInteger();
logWriter = null;
flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging;
singlePageFlushStrategy = this::flushSinglePageWithoutLogging;
try {
if (flushWithLogging) {
// without RATIS, utilize physical logging for integrity
int pageAcc = (int) recoverFromLog(logPath) / SchemaFileConfig.PAGE_LENGTH;
this.logWriter = new SchemaFileLogWriter(logPath);
logCounter = new AtomicInteger(pageAcc);
flushDirtyPagesStrategy = this::flushDirtyPagesWithLogging;
singlePageFlushStrategy = this::flushSinglePageWithLogging;
} else {
// with RATIS enabled, integrity is guaranteed by consensus protocol
logCounter = new AtomicInteger();
logWriter = null;
flushDirtyPagesStrategy = this::flushDirtyPagesWithoutLogging;
singlePageFlushStrategy = this::flushSinglePageWithoutLogging;
}
} catch (IOException | MetadataException e) {
try {
close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
}

Expand All @@ -75,28 +84,57 @@ public void renewLogWriter() throws IOException {
}
}

public void closeLogWriter() throws IOException {
public void close() throws IOException {
IOException failure = null;
if (logWriter != null) {
logWriter.close();
try {
logWriter.close();
} catch (IOException e) {
failure = e;
}
}
try {
readChannel.close();
} catch (IOException e) {
if (failure == null) {
failure = e;
} else {
failure.addSuppressed(e);
}
}
if (failure != null) {
throw failure;
}
}

/** Load bytes from log, deserialize and flush directly into channel, return current length */
private long recoverFromLog(String logPath) throws IOException, MetadataException {
SchemaFileLogReader reader = new SchemaFileLogReader(logPath);
ISchemaPage page;
List<byte[]> res = reader.collectUpdatedEntries();
for (byte[] entry : res) {
// TODO check bytes semantic correctness with CRC32 or other way
page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry));
page.flushPageToChannel(this.channel);
List<byte[]> res;
try {
res = reader.collectUpdatedEntries();
for (byte[] entry : res) {
// Check byte semantics with CRC32 or another method.
page = ISchemaPage.loadSchemaPage(ByteBuffer.wrap(entry));
page.flushPageToChannel(this.channel);
}
} catch (IOException | MetadataException e) {
try {
reader.close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
reader.close();

// complete log file
if (!res.isEmpty()) {
try (FileOutputStream outputStream = new FileOutputStream(logPath, true)) {
outputStream.write(new byte[] {SchemaFileConfig.SF_COMMIT_MARK});
IOUtils.writeFully(
outputStream.getChannel(),
ByteBuffer.wrap(new byte[] {SchemaFileConfig.SF_COMMIT_MARK}));
return outputStream.getChannel().size();
}
}
Expand Down
27 changes: 18 additions & 9 deletions .../iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,14 +85,23 @@ public abstract class PageManager implements IPageManager {
pageIOChannel = new PageIOChannel(channel, pmtFile, true, logPath);
}

// construct first page if file to init
if (lastPageIndex < 0) {
ISegmentedPage rootPage =
ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0);
rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ);
pagePool.put(rootPage);
rootPage.syncPageBuffer();
rootPage.flushPageToChannel(channel);
try {
// construct first page if file to init
if (lastPageIndex < 0) {
ISegmentedPage rootPage =
ISchemaPage.initSegmentedPage(ByteBuffer.allocate(SchemaFileConfig.PAGE_LENGTH), 0);
rootPage.allocNewSegment(SchemaFileConfig.SEG_MAX_SIZ);
pagePool.put(rootPage);
rootPage.syncPageBuffer();
rootPage.flushPageToChannel(channel);
}
} catch (IOException | MetadataException e) {
try {
pageIOChannel.close();
} catch (IOException closeException) {
e.addSuppressed(closeException);
}
throw e;
}
}

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

@Override
public void close() throws IOException {
pageIOChannel.closeLogWriter();
pageIOChannel.close();
}

// endregion
Expand Down
Loading
Loading