You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/06/28 07:47:45 UTC
[GitHub] [flink-table-store] JingsongLi opened a new pull request, #178: [FLINK-28244] Introduce changelog file for DataFile
JingsongLi opened a new pull request, #178:
URL: https://github.com/apache/flink-table-store/pull/178
When using TableStore to support stream consumption, there are two requirements.
- Downstream gets all changelogs
- The order of stream consumption is the order of input
For append only table, it is easy to meet both.
But for the primary key table, Its files are all sorted and de-duplicated by pk, making it impossible to meet the above expectations.
We can output another ChangelogFile when the DataFile flush, and the stream reads it directly.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] tsreaper commented on a diff in pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
tsreaper commented on code in PR #178:
URL: https://github.com/apache/flink-table-store/pull/178#discussion_r910559342
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/DataFileWriter.java:
##########
@@ -105,32 +106,60 @@ public DataFilePathFactory pathFactory() {
return pathFactory;
}
+ /** Write raw {@link KeyValue} iterator into a changelog file. */
+ public Path writeLevel0Changelog(Iterator<KeyValue> iterator) throws Exception {
+ FileWriter.Factory<KeyValue, Metric> writerFactory = createFileWriterFactory();
+ Path changelogPath = pathFactory.newChangelogPath();
+ doWrite(writerFactory.create(changelogPath), iterator);
+ return changelogPath;
+ }
+
+ /**
+ * Write several {@link KeyValue}s into a data file of level 0.
+ *
+ * @return empty if iterator is empty
+ */
+ public Optional<DataFileMeta> writeLevel0(Iterator<KeyValue> iterator) throws Exception {
+ List<DataFileMeta> files = write(iterator, 0);
+ if (files.size() > 1) {
+ throw new RuntimeException("Produce illegal multiple Level 0 files: " + files);
+ }
+ return files.size() == 0 ? Optional.empty() : Optional.of(files.get(0));
+ }
+
/**
- * Write several {@link KeyValue}s into a data file of a given level.
+ * Write several {@link KeyValue}s into data files of a given level.
*
* <p>NOTE: This method is atomic.
*/
- public List<DataFileMeta> write(CloseableIterator<KeyValue> iterator, int level)
- throws Exception {
+ public List<DataFileMeta> write(Iterator<KeyValue> iterator, int level) throws Exception {
+ // Don't roll file for level 0
+ long suggestedFileSize = level == 0 ? Long.MAX_VALUE : this.suggestedFileSize;
+ return doWrite(createRollingKvWriter(level, suggestedFileSize), iterator);
+ }
- RollingKvWriter rollingKvWriter = createRollingKvWriter(level, suggestedFileSize);
- try (RollingKvWriter writer = rollingKvWriter) {
+ private <R> R doWrite(FileWriter<KeyValue, R> fileWriter, Iterator<KeyValue> iterator)
+ throws Exception {
+ try (FileWriter<KeyValue, R> writer = fileWriter) {
writer.write(iterator);
-
} catch (Throwable e) {
LOG.warn("Exception occurs when writing data files. Cleaning up.", e);
-
- rollingKvWriter.abort();
+ fileWriter.abort();
throw e;
} finally {
- iterator.close();
+ if (iterator instanceof AutoCloseable) {
+ ((AutoCloseable) iterator).close();
Review Comment:
I mean we can remove this `instanceof` checking if we always provide a `ClosableIterator`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a diff in pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on code in PR #178:
URL: https://github.com/apache/flink-table-store/pull/178#discussion_r909620472
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/DataFileWriter.java:
##########
@@ -105,32 +106,60 @@ public DataFilePathFactory pathFactory() {
return pathFactory;
}
+ /** Write raw {@link KeyValue} iterator into a changelog file. */
+ public Path writeLevel0Changelog(Iterator<KeyValue> iterator) throws Exception {
+ FileWriter.Factory<KeyValue, Metric> writerFactory = createFileWriterFactory();
+ Path changelogPath = pathFactory.newChangelogPath();
+ doWrite(writerFactory.create(changelogPath), iterator);
+ return changelogPath;
+ }
+
+ /**
+ * Write several {@link KeyValue}s into a data file of level 0.
+ *
+ * @return empty if iterator is empty
+ */
+ public Optional<DataFileMeta> writeLevel0(Iterator<KeyValue> iterator) throws Exception {
+ List<DataFileMeta> files = write(iterator, 0);
+ if (files.size() > 1) {
+ throw new RuntimeException("Produce illegal multiple Level 0 files: " + files);
+ }
+ return files.size() == 0 ? Optional.empty() : Optional.of(files.get(0));
+ }
+
/**
- * Write several {@link KeyValue}s into a data file of a given level.
+ * Write several {@link KeyValue}s into data files of a given level.
*
* <p>NOTE: This method is atomic.
*/
- public List<DataFileMeta> write(CloseableIterator<KeyValue> iterator, int level)
- throws Exception {
+ public List<DataFileMeta> write(Iterator<KeyValue> iterator, int level) throws Exception {
+ // Don't roll file for level 0
+ long suggestedFileSize = level == 0 ? Long.MAX_VALUE : this.suggestedFileSize;
+ return doWrite(createRollingKvWriter(level, suggestedFileSize), iterator);
+ }
- RollingKvWriter rollingKvWriter = createRollingKvWriter(level, suggestedFileSize);
- try (RollingKvWriter writer = rollingKvWriter) {
+ private <R> R doWrite(FileWriter<KeyValue, R> fileWriter, Iterator<KeyValue> iterator)
+ throws Exception {
+ try (FileWriter<KeyValue, R> writer = fileWriter) {
writer.write(iterator);
-
} catch (Throwable e) {
LOG.warn("Exception occurs when writing data files. Cleaning up.", e);
-
- rollingKvWriter.abort();
+ fileWriter.abort();
throw e;
} finally {
- iterator.close();
+ if (iterator instanceof AutoCloseable) {
+ ((AutoCloseable) iterator).close();
Review Comment:
Yes, `ClosableIterator` is a `Iterator` with `AutoCloseable`.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi closed pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
JingsongLi closed pull request #178: [FLINK-28244] Introduce changelog file for DataFile
URL: https://github.com/apache/flink-table-store/pull/178
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi merged pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
JingsongLi merged PR #178:
URL: https://github.com/apache/flink-table-store/pull/178
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] tsreaper commented on a diff in pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
tsreaper commented on code in PR #178:
URL: https://github.com/apache/flink-table-store/pull/178#discussion_r909398296
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/data/DataFileWriter.java:
##########
@@ -105,32 +106,60 @@ public DataFilePathFactory pathFactory() {
return pathFactory;
}
+ /** Write raw {@link KeyValue} iterator into a changelog file. */
+ public Path writeLevel0Changelog(Iterator<KeyValue> iterator) throws Exception {
+ FileWriter.Factory<KeyValue, Metric> writerFactory = createFileWriterFactory();
+ Path changelogPath = pathFactory.newChangelogPath();
+ doWrite(writerFactory.create(changelogPath), iterator);
+ return changelogPath;
+ }
+
+ /**
+ * Write several {@link KeyValue}s into a data file of level 0.
+ *
+ * @return empty if iterator is empty
+ */
+ public Optional<DataFileMeta> writeLevel0(Iterator<KeyValue> iterator) throws Exception {
+ List<DataFileMeta> files = write(iterator, 0);
+ if (files.size() > 1) {
+ throw new RuntimeException("Produce illegal multiple Level 0 files: " + files);
+ }
+ return files.size() == 0 ? Optional.empty() : Optional.of(files.get(0));
+ }
+
/**
- * Write several {@link KeyValue}s into a data file of a given level.
+ * Write several {@link KeyValue}s into data files of a given level.
*
* <p>NOTE: This method is atomic.
*/
- public List<DataFileMeta> write(CloseableIterator<KeyValue> iterator, int level)
- throws Exception {
+ public List<DataFileMeta> write(Iterator<KeyValue> iterator, int level) throws Exception {
+ // Don't roll file for level 0
+ long suggestedFileSize = level == 0 ? Long.MAX_VALUE : this.suggestedFileSize;
+ return doWrite(createRollingKvWriter(level, suggestedFileSize), iterator);
+ }
- RollingKvWriter rollingKvWriter = createRollingKvWriter(level, suggestedFileSize);
- try (RollingKvWriter writer = rollingKvWriter) {
+ private <R> R doWrite(FileWriter<KeyValue, R> fileWriter, Iterator<KeyValue> iterator)
+ throws Exception {
+ try (FileWriter<KeyValue, R> writer = fileWriter) {
writer.write(iterator);
-
} catch (Throwable e) {
LOG.warn("Exception occurs when writing data files. Cleaning up.", e);
-
- rollingKvWriter.abort();
+ fileWriter.abort();
throw e;
} finally {
- iterator.close();
+ if (iterator instanceof AutoCloseable) {
+ ((AutoCloseable) iterator).close();
Review Comment:
We can still accept `ClosableIterator` as parameter. Raw iterator just does nothing when closed.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeWriter.java:
##########
@@ -138,11 +142,28 @@ public void flushMemory() throws Exception {
// stop writing, wait for compaction finished
finishCompaction(true);
}
- Iterator<KeyValue> iterator = memTable.iterator(keyComparator, mergeFunction);
- List<DataFileMeta> files =
- dataFileWriter.write(CloseableIterator.adapterForIterator(iterator), 0);
- newFiles.addAll(files);
- files.forEach(levels::addLevel0File);
+ List<String> extraFiles = new ArrayList<>();
+ if (enableChangelogFile) {
+ extraFiles.add(
+ dataFileWriter.writeLevel0Changelog(memTable.rawIterator()).getName());
+ }
+ AtomicBoolean success = new AtomicBoolean(false);
+ try {
+ Iterator<KeyValue> iterator = memTable.mergeIterator(keyComparator, mergeFunction);
+ dataFileWriter
+ .writeLevel0(iterator)
+ .ifPresent(
+ file -> {
+ DataFileMeta fileMeta = file.copy(extraFiles);
+ newFiles.add(fileMeta);
+ levels.addLevel0File(fileMeta);
+ success.set(true);
+ });
+ } finally {
+ if (!success.get()) {
+ extraFiles.forEach(dataFileWriter::delete);
+ }
+ }
Review Comment:
```suggestion
boolean success = false;
try {
Iterator<KeyValue> iterator = memTable.mergeIterator(keyComparator, mergeFunction);
success =
dataFileWriter
.writeLevel0(iterator)
.map(
file -> {
DataFileMeta fileMeta = file.copy(extraFiles);
newFiles.add(fileMeta);
levels.addLevel0File(fileMeta);
return true;
})
.orElse(false);
} finally {
if (!success) {
extraFiles.forEach(dataFileWriter::delete);
}
}
```
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi closed pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
JingsongLi closed pull request #178: [FLINK-28244] Introduce changelog file for DataFile
URL: https://github.com/apache/flink-table-store/pull/178
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi closed pull request #178: [FLINK-28244] Introduce changelog file for DataFile
Posted by GitBox <gi...@apache.org>.
JingsongLi closed pull request #178: [FLINK-28244] Introduce changelog file for DataFile
URL: https://github.com/apache/flink-table-store/pull/178
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org