Bound total memory used by HiveSplitSource #9119
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
electrum
approved these changes
Oct 9, 2017
| CompletableFuture<List<ConnectorSplit>> future = queue.getBatchAsync(maxSize); | ||
| CompletableFuture<List<ConnectorSplit>> future = queue.getBatchAsync(maxSize).thenApply(internalSplits -> { | ||
| ImmutableList.Builder<ConnectorSplit> result = ImmutableList.builder(); | ||
| for (InternalHiveSplit internalSplit : internalSplits) { |
There was a problem hiding this comment.
The next commit introduces side effect to the for-loop
| public final class HiveTypeName | ||
| { | ||
| private static final int INSTANCE_SIZE = ClassLayout.parseClass(HivePartitionKey.class).instanceSize() + | ||
| ClassLayout.parseClass(String.class).instanceSize() * 2; |
There was a problem hiding this comment.
Why times two? There is only one String in the class.
| this.value = requireNonNull(value, "value is null"); | ||
| } | ||
|
|
||
| public String toString() |
| @@ -264,7 +264,7 @@ public HiveWriter createWriter(Page partitionColumns, int position, OptionalInt | |||
| .collect(joining(","))); | |||
| schema.setProperty(META_TABLE_COLUMN_TYPES, dataColumns.stream() | |||
| .map(DataColumn::getHiveType) | |||
| .map(HiveType::getHiveTypeName) | |||
| .map(hiveType -> hiveType.getHiveTypeName().toString()) | |||
There was a problem hiding this comment.
Just add
.map(Object::toString)| @@ -463,7 +462,7 @@ private static Properties createSchema(HiveStorageFormat format, List<String> co | |||
| .collect(joining(","))); | |||
| schema.setProperty(META_TABLE_COLUMN_TYPES, columnTypes.stream() | |||
| .map(type -> toHiveType(typeTranslator, type)) | |||
| .map(HiveType::getHiveTypeName) | |||
| .map(hiveType -> hiveType.getHiveTypeName().toString()) | |||
| @@ -51,6 +55,11 @@ public String getValue() | |||
| return value; | |||
| } | |||
|
|
|||
| public int getEstimatedSizeInBytes() | |||
| { | |||
| return INSTANCE_SIZE + name.length() * Character.BYTES + value.length() * Character.BYTES; | |||
There was a problem hiding this comment.
This is long and hard to read, add parentheses
There was a problem hiding this comment.
I disagree on this one. I find it harder to read with parenthesis.
I will address all other comments. I noticed the build failure and fixed them myself. Apparently, I forgot to push it.
| @@ -62,6 +73,7 @@ | |||
| this.tableName = requireNonNull(tableName, "tableName is null"); | |||
| this.compactEffectivePredicate = requireNonNull(compactEffectivePredicate, "compactEffectivePredicate is null"); | |||
| this.queue = new AsyncQueue<>(maxOutstandingSplits, executor); | |||
| this.maxOutstandingSplitsBytes = Ints.checkedCast(maxOutstandingSplitsSize.toBytes()); | |||
There was a problem hiding this comment.
Using Ints.saturatedCast() might be better here. It means the memory limit will be lower than the very large limit that was specified.
| @@ -84,9 +96,16 @@ int getOutstandingSplitCount() | |||
| CompletableFuture<?> addToQueue(InternalHiveSplit split) | |||
| { | |||
| if (throwable.get() == null) { | |||
| if (estimatedSplitSizeInBytes.addAndGet(split.getEstimatedSizeInBytes()) > maxOutstandingSplitsBytes) { | |||
There was a problem hiding this comment.
This can overflow if the limit is close to MAX_INT, and/or if there are many threads incrementing at once
There was a problem hiding this comment.
We should probably make this a long, just to be safe
| // This limit should never be hit given there is a limit of maxOutstandingSplits. | ||
| // If it's hit, it means individual splits are huge. | ||
| throw new PrestoException(GENERIC_INTERNAL_ERROR, format( | ||
| "Split buffering for %s.%s takes too much memory (%s bytes limit). %s splits are buffered.", |
There was a problem hiding this comment.
format(
"Split buffering for %s.%s exceeded memory limit (%s). %s splits are buffered.",
..., succinctBytes(maxOutstandingSplitsBytes), ...)
haozhun
force-pushed
the
hive-split
branch
4 times, most recently
from
924793b
to
51fc023
Compare
InternalHiveSplit avoids unnecessary duplicate information, and is more friendly to memory accounting. It is used for buffering discovered splits inside Hive connector.
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.
No description provided.