You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "JackieTien97 (via GitHub)" <gi...@apache.org> on 2023/07/27 02:14:16 UTC
[GitHub] [iotdb] JackieTien97 commented on a diff in pull request #10686: Implement stream interface of Mods read
JackieTien97 commented on code in PR #10686:
URL: https://github.com/apache/iotdb/pull/10686#discussion_r1275649923
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/io/LocalTextModificationAccessor.java:
##########
@@ -114,23 +149,68 @@ public void close() throws IOException {
}
@Override
- public void abort() throws IOException {
+ public void write(Modification mod) throws IOException {
if (writer == null) {
writer = FSFactoryProducer.getFSFactory().getBufferedWriter(filePath, true);
}
- writer.write(ABORT_MARK);
+ writer.write(encodeModification(mod));
writer.newLine();
writer.flush();
}
- @Override
- public void write(Modification mod) throws IOException {
+ @TestOnly
+ public void writeInComplete(Modification mod) throws IOException {
if (writer == null) {
writer = FSFactoryProducer.getFSFactory().getBufferedWriter(filePath, true);
}
- writer.write(encodeModification(mod));
- writer.newLine();
- writer.flush();
+ writer.write(encodeModification(mod).substring(0, 2));
Review Comment:
deal with possible NPE
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2005,39 +1986,47 @@ private void deleteDataInFiles(
}
ModificationFile modFile = tsFileResource.getModFile();
+ long originSize = modFile.getSize();
Review Comment:
```suggestion
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2005,39 +1986,47 @@ private void deleteDataInFiles(
}
ModificationFile modFile = tsFileResource.getModFile();
+ long originSize = modFile.getSize();
if (tsFileResource.isClosed()) {
- // delete data in sealed file
- if (tsFileResource.isCompacting()) {
- // we have to set modification offset to MAX_VALUE, as the offset of source chunk may
- // change after compaction
- deletion.setFileOffset(Long.MAX_VALUE);
- // write deletion into compaction modification file
- tsFileResource.getCompactionModFile().write(deletion);
- // write deletion into modification file to enable read during compaction
- modFile.write(deletion);
- // remember to close mod file
- tsFileResource.getCompactionModFile().close();
- modFile.close();
- } else {
- deletion.setFileOffset(tsFileResource.getTsFileSize());
- // write deletion into modification file
- boolean modFileExists = modFile.exists();
- long originSize = modFile.getSize();
- modFile.write(deletion);
+ try {
+ tsFileResource.writeLock();
Review Comment:
```suggestion
tsFileResource.writeLock();
originSize = modFile.getSize();
```
We need to get file length inside lock, otherwise, there will be concurrent problems. ThreadB may truncate ThreadA's complete mod entry.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2005,39 +1986,47 @@ private void deleteDataInFiles(
}
ModificationFile modFile = tsFileResource.getModFile();
+ long originSize = modFile.getSize();
if (tsFileResource.isClosed()) {
- // delete data in sealed file
- if (tsFileResource.isCompacting()) {
- // we have to set modification offset to MAX_VALUE, as the offset of source chunk may
- // change after compaction
- deletion.setFileOffset(Long.MAX_VALUE);
- // write deletion into compaction modification file
- tsFileResource.getCompactionModFile().write(deletion);
- // write deletion into modification file to enable read during compaction
- modFile.write(deletion);
- // remember to close mod file
- tsFileResource.getCompactionModFile().close();
- modFile.close();
- } else {
- deletion.setFileOffset(tsFileResource.getTsFileSize());
- // write deletion into modification file
- boolean modFileExists = modFile.exists();
- long originSize = modFile.getSize();
- modFile.write(deletion);
+ try {
+ tsFileResource.writeLock();
+ // delete data in sealed file
+ if (tsFileResource.isCompacting()) {
+ // we have to set modification offset to MAX_VALUE, as the offset of source chunk may
+ // change after compaction
+ deletion.setFileOffset(Long.MAX_VALUE);
+ // write deletion into compaction modification file
+ tsFileResource.getCompactionModFile().write(deletion);
+ // write deletion into modification file to enable read during compaction
+ modFile.write(deletion);
+ // remember to close mod file
+ tsFileResource.getCompactionModFile().close();
+ modFile.close();
+ } else {
+ deletion.setFileOffset(tsFileResource.getTsFileSize());
+ // write deletion into modification file
+ boolean modFileExists = modFile.exists();
- // remember to close mod file
- modFile.close();
+ modFile.write(deletion);
- // if file length greater than 1M,execute compact.
- modFile.compact();
+ // remember to close mod file
+ modFile.close();
- if (!modFileExists) {
- FileMetrics.getInstance().increaseModFileNum(1);
- }
+ // if file length greater than 1M,execute compact.
+ modFile.compact();
+
+ if (!modFileExists) {
+ FileMetrics.getInstance().increaseModFileNum(1);
+ }
- // The file size may be smaller than the original file, so the increment here may be
- // negative
- FileMetrics.getInstance().increaseModFileSize(modFile.getSize() - originSize);
+ // The file size may be smaller than the original file, so the increment here may be
+ // negative
+ FileMetrics.getInstance().increaseModFileSize(modFile.getSize() - originSize);
+ }
+ } catch (IOException e) {
+ modFile.truncate(originSize);
Review Comment:
```suggestion
} catch (Throwable t) {
if (originSize != -1) {
modFile.truncate(originSize);
}
throw t;
```
We must rethrow the exception, otherwise this excption will be ignored, we should at least let user know that their deletion operation failed.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/io/LocalTextModificationAccessor.java:
##########
@@ -65,44 +70,74 @@ public LocalTextModificationAccessor(String filePath) {
@Override
public Collection<Modification> read() {
+ List<Modification> result = new ArrayList<>();
+ Iterator<Modification> iterator = getModificationIterator();
+ while (iterator.hasNext()) {
+ result.add(iterator.next());
+ }
+ return result;
+ }
+
+ @Override
+ public Iterator<Modification> getModificationIterator() {
File file = FSFactoryProducer.getFSFactory().getFile(filePath);
- if (!file.exists()) {
- logger.debug("No modification has been written to this file");
- return new ArrayList<>();
- }
-
- String line;
- long truncatedSize = 0;
- boolean crashed = false;
- List<Modification> modificationList = new ArrayList<>();
- try (TracedBufferedReader reader = new TracedBufferedReader(new FileReader(file))) {
- while ((line = reader.readLine()) != null) {
- if (line.equals(ABORT_MARK) && !modificationList.isEmpty()) {
- modificationList.remove(modificationList.size() - 1);
- } else {
- modificationList.add(decodeModification(line));
+ final BufferedReader reader;
+ try {
+ reader = new BufferedReader(new FileReader(file));
+ } catch (FileNotFoundException e) {
+ logger.debug(NO_MODIFICATION_MSG, file);
+
+ // return empty iterator
+ return new Iterator<Modification>() {
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Modification next() {
+ throw new NoSuchElementException();
+ }
+ };
+ }
+
+ final Modification[] cachedModification = new Modification[1];
+ return new Iterator<Modification>() {
+ @Override
+ public boolean hasNext() {
+ try {
+ if (cachedModification[0] == null) {
+ String line = reader.readLine();
+ if (line == null) {
+ reader.close();
+ return false;
+ } else {
+ try {
+ cachedModification[0] = decodeModification(line);
+ } catch (IOException e) {
+ logger.warn("An error occurred when decode line-[{}] to modification", line);
+ cachedModification[0] = null;
+ reader.close();
+ return false;
+ }
Review Comment:
Try-catch blocks should not be nested
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java:
##########
@@ -2005,39 +1986,47 @@ private void deleteDataInFiles(
}
ModificationFile modFile = tsFileResource.getModFile();
+ long originSize = modFile.getSize();
if (tsFileResource.isClosed()) {
- // delete data in sealed file
- if (tsFileResource.isCompacting()) {
- // we have to set modification offset to MAX_VALUE, as the offset of source chunk may
- // change after compaction
- deletion.setFileOffset(Long.MAX_VALUE);
- // write deletion into compaction modification file
- tsFileResource.getCompactionModFile().write(deletion);
- // write deletion into modification file to enable read during compaction
- modFile.write(deletion);
- // remember to close mod file
- tsFileResource.getCompactionModFile().close();
- modFile.close();
- } else {
- deletion.setFileOffset(tsFileResource.getTsFileSize());
- // write deletion into modification file
- boolean modFileExists = modFile.exists();
- long originSize = modFile.getSize();
- modFile.write(deletion);
+ try {
Review Comment:
```suggestion
long originSize = -1;
try {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org