Make UDTF execution framework batch processing #12562
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
lancelly
reviewed
May 27, 2024
| } | ||
|
|
||
| Column[] columns = transformers[0].current(); | ||
| TsBlock block = new TsBlock((TimeColumn) columns[1], columns[0]); |
There was a problem hiding this comment.
Why not use API wrapBlocksWithoutCopy🤔？
There was a problem hiding this comment.
With this API:
int count = columns[0].getPositionCount();
Column[] valueColumns = { columns[0] };
TsBlock block = TsBlock.wrapBlocksWithoutCopy(count, (TimeColumn) columns[1], valueColumns);Without this API:
TsBlock block = new TsBlock((TimeColumn) columns[1], columns[0]);The code hasn't gotten any more elegant.
lancelly
reviewed
May 27, 2024
| } | ||
|
|
||
| Column[] columns = transformers[index].current(); | ||
| TsBlock block = new TsBlock((TimeColumn) columns[1], columns[0]); |
lancelly
reviewed
May 27, 2024
|
|
||
| public ConstantInputReader(ConstantOperand expression) throws QueryProcessException { | ||
| this.expression = Validate.notNull(expression); | ||
| Validate.notNull(expression); |
There was a problem hiding this comment.
This is a Deprecated method. You could add some error msg.
lancelly
reviewed
May 27, 2024
| public boolean currentBoolean() throws IOException { | ||
| return cachedBoolean; | ||
| public Column[] current() throws IOException { | ||
| return new Column[] {cachedColumn}; |
There was a problem hiding this comment.
Perhaps making cachedColumn Column[] instead of Column could avoid new array each time.
lancelly
reviewed
May 27, 2024
|
|
||
| public class QueryDataSetInputLayer { | ||
|
|
||
| private IUDFInputDataSet queryDataSet; | ||
| private TsBlockInputDataSet queryDataSet; |
There was a problem hiding this comment.
Perhaps using interface IUDFInputDataSet is better?
There was a problem hiding this comment.
I think TsBlockInputDataSet would provide more information than IUDFInputDataSet .
And QueryDataSetInputLayer must be TsBlockInputDataSet.
lancelly
reviewed
May 27, 2024
| cachedRowRecord = null; | ||
| currentRowIndex = -1; | ||
| } | ||
| private RowListForwardIterator iterator = rowList.constructIterator(); |
There was a problem hiding this comment.
private final RowListForwardIterator iterator
lancelly
reviewed
May 29, 2024
|
|
||
| void readyForNext(); | ||
| void consumed(int count); |
lancelly
reviewed
May 29, 2024
|
|
||
| executor.execute(columns, valueColumnBuilder); | ||
|
|
||
| Column timeColumn = columns[1]; |
There was a problem hiding this comment.
Make sure that columns returned by layerReader only contains two columns.
lancelly
reviewed
May 29, 2024
| } | ||
|
|
||
| public long getTimeByIndex(int index) { | ||
| assert !isConstant; |
There was a problem hiding this comment.
throw UnsupportedException instead
There was a problem hiding this comment.
Actually TVColumns class is only for internal use. So it will never trigger the assertion failed.
I don't think there is the need to throw an exception.
lancelly
reviewed
May 29, 2024
| // region single data methods | ||
| public long getTime(int index) { | ||
| // Never access null row's time | ||
| assert index >= prefixNullCount; |
There was a problem hiding this comment.
replace assert with checkState()
There was a problem hiding this comment.
What are the benefits of the checkState() over the asset?
|
PTAL~ |
SzyWilliam
pushed a commit
to SzyWilliam/iotdb
that referenced
this pull request
Nov 26, 2024
* Add `.putFloats()` to insert multiple rows at the same time. * Complete all batch insert method in `BatchData.java`. * Add overload methods with indexes to avoid copying. * Add Java doc for new methods in `BatchData.java`. * Merge two `BatchDataTest.java` files. * Add `.putColumns()` method to `SerializableRowRecordList.java`. * Finish batch insert method prototype in row record list and tv list. * UDTF execution framework can support running really simple UDTF. * Change TVList to list of columns to avoid copying. * Compile DataNode without error. * Format all files by `mvn splotless:apply`. * Optimize import by using IDEA shortcut. * Fix put data into existing list error. * Generate data in terminate method. * Format again using `mvn spotless:apply`. * Fix some bugs and implement RowWindow UDTF. * Basically pass all UDF-related IT. * Add license header to all newly-added files. * Add memory control strategy for binary list. * Add memory control strategy for row list that contains binary field. * Copy subregion to avoid memory leakage. * Update iterator by recording first data index when apply new memory strategy. * Unify `newXXX` series methods name to `construct`. * Basically refactor storage layer. * Add and update copy method to Column based classes. * Rewrite unit test for ElasticSerializableRowList. * Rewrite unit test for SerializableRowList. * Rewrite unit test for SerializableTVList. * Rewrite unit test for ElasticSerializableTVList. * Sync changes between TVList and RowList. * Refine UT of ESRowList and ESTVList with random size strings. * Do not cache internal list in some situation. * Add forgotten files. * Revert `BatchData.java` relevant files. * Remove redundant `timeHeap.clear()` in MultiInputColumnIntermediateLayer. * Add constant folding in BinaryTransformer. * Refactor computational layer. * Replace origin normal column in constant reader with RLEColumn. * Batch sliding time row window in SingleInputColumnSingleReferenceIntermediateLayer. * Batch other row windows in SingleInputColumnSingleReferenceIntermediateLayer. * Add forgotten `TransformUtils.java` file. * Batch all row windows in SingleInputColumnMultiReferenceIntermediateLayer. * Batch all row windows in all intermediate layer. * Remove code with no usage. * Remove code with no usage agaion. * Remove nonexistent files. * Refactor project's structure. * Optimize when there is only one transformer. * Fix losing data bug in benchmarking. * Refactor UDTF executor. * Skip cache update when getting column size. * Discard TsBlock when there is only one UDF. * Fix transformer IT errors. * Remove unnecessary bitmap in ESRowList. * Update TsFile version. * Fix some IT OOM bugs. * Try to fix UT OOM error. * Apply spotless plugin. * Evict data in UDF output storage. * Revert optimization when there is only one UDTF. * Fix losing data error. * Update internal column count when ESList apply new memory control strategy. * Add new types in SerializableRowList. * Apply mvn spotless plugin. * Fix integration-tests errors. * Modify code according to committer's advices.[Part One] * Modify code according to committer's advices.[Part Two] * Modify code according to committer's advices.[Part Three]
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Description
The origin UDTF execution framework process one data point at time, which is inefficient. This PR introduce a new UDTF execution framework which process one batch at time. While lowering the function call times, it also reduce the copy significantly.
Additionally, this PR also introduce some optimizations like constant folding and skip time join when there is only one UDTF expression. The code is refactor to improve readability as well.
This PR has:
for an unfamiliar reader.