924060929 commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4449511697
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileSplitter.java:
##########
@@ -87,132 +86,112 @@ public FileSplitter(long maxInitialSplitSize, long
maxSplitSize, int maxInitialS
this.maxInitialSplitSize = maxInitialSplitSize;
this.maxSplitSize = maxSplitSize;
this.maxInitialSplitNum = maxInitialSplitNum;
- currentMaxSplitSize = maxInitialSplitSize;
remainingInitialSplitNum = new AtomicInteger(maxInitialSplitNum);
}
- public List<Split> splitFile(
- LocationPath path,
- long specifiedFileSplitSize,
- BlockLocation[] blockLocations,
- long length,
- long modificationTime,
- boolean splittable,
- List<String> partitionValues,
- SplitCreator splitCreator)
- throws IOException {
- if (length <= 0) {
- // Zero-length files contain no data; skip to avoid sending empty
splits to BE.
- return Lists.newArrayList();
- }
- // Pass splitCreator.create() to set target file split size to
calculate split weight.
- long targetFileSplitSize = specifiedFileSplitSize > 0 ?
specifiedFileSplitSize : maxSplitSize;
- if (blockLocations == null) {
- blockLocations = new BlockLocation[1];
- blockLocations[0] = new BlockLocation(null, null, 0L, length);
- }
+ public List<Split> splitFile(LocationPath path, long
specifiedFileSplitSize,
+ BlockLocation[] blockLocations, long length, long
modificationTime, boolean splittable,
+ List<String> partitionValues, SplitCreator splitCreator) throws
IOException {
List<Split> result = Lists.newArrayList();
- TFileCompressType compressType =
Util.inferFileCompressTypeByPath(path.getNormalizedLocation());
- if (!splittable || compressType != TFileCompressType.PLAIN) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Path {} is not splittable.", path);
- }
- String[] hosts = blockLocations.length == 0 ? null :
blockLocations[0].getHosts();
- result.add(splitCreator.create(path, 0, length, length,
- targetFileSplitSize,
- modificationTime, hosts, partitionValues));
- updateCurrentMaxSplitSize();
- return result;
+ try {
+ splitFileIterator(path, specifiedFileSplitSize, blockLocations,
length, modificationTime,
+ splittable, partitionValues,
splitCreator).forEachRemaining(result::add);
+ } catch (UncheckedIOException e) {
+ throw e.getCause();
}
+ return result;
+ }
- // if specified split size is not zero, split file by specified size
- if (specifiedFileSplitSize > 0) {
- long bytesRemaining;
- for (bytesRemaining = length; (double) bytesRemaining / (double)
specifiedFileSplitSize > 1.1D;
- bytesRemaining -= specifiedFileSplitSize) {
- int location = getBlockIndex(blockLocations, length -
bytesRemaining);
- String[] hosts = location == -1 ? null :
blockLocations[location].getHosts();
- result.add(splitCreator.create(path, length - bytesRemaining,
specifiedFileSplitSize,
- length, specifiedFileSplitSize, modificationTime,
hosts, partitionValues));
- }
- if (bytesRemaining != 0L) {
- int location = getBlockIndex(blockLocations, length -
bytesRemaining);
- String[] hosts = location == -1 ? null :
blockLocations[location].getHosts();
- result.add(splitCreator.create(path, length - bytesRemaining,
bytesRemaining,
- length, specifiedFileSplitSize, modificationTime,
hosts, partitionValues));
- }
- return result;
+ /** Uses the existing file boundaries while constructing only ranges
actually consumed. */
+ public Iterator<Split> splitFileIterator(LocationPath path, long
specifiedFileSplitSize,
+ BlockLocation[] blockLocations, long length, long
modificationTime, boolean splittable,
+ List<String> partitionValues, SplitCreator splitCreator) throws
IOException {
+ if (length <= 0) {
+ return Collections.emptyIterator();
}
-
- // split file by block
- long start = 0;
+ long targetFileSplitSize = specifiedFileSplitSize > 0 ?
specifiedFileSplitSize : maxSplitSize;
+ BlockLocation[] locations = blockLocations == null
+ ? new BlockLocation[] {new BlockLocation(null, null, 0L,
length)} : blockLocations;
+ boolean canSplit = splittable
+ &&
Util.inferFileCompressTypeByPath(path.getNormalizedLocation()) ==
TFileCompressType.PLAIN;
ImmutableList.Builder<InternalBlock> blockBuilder =
ImmutableList.builder();
- for (BlockLocation blockLocation : blockLocations) {
- // clamp the block range
- long blockStart = Math.max(start, blockLocation.getOffset());
- long blockEnd = Math.min(start + length, blockLocation.getOffset()
+ blockLocation.getLength());
- if (blockStart > blockEnd) {
- // block is outside split range
- continue;
- }
- if (blockStart == blockEnd && !(blockStart == start && blockEnd ==
start + length)) {
- // skip zero-width block, except in the special circumstance:
- // slice is empty, and the block covers the empty slice
interval.
- continue;
+ if (canSplit && specifiedFileSplitSize <= 0) {
+ for (BlockLocation block : locations) {
+ long blockStart = Math.max(0, block.getOffset());
+ long blockEnd = Math.min(length, block.getOffset() +
block.getLength());
+ if (blockStart > blockEnd || blockStart == blockEnd && length
> 0) {
+ continue;
+ }
+ blockBuilder.add(new InternalBlock(blockStart, blockEnd,
block.getHosts()));
}
- blockBuilder.add(new InternalBlock(blockStart, blockEnd,
blockLocation.getHosts()));
}
List<InternalBlock> blocks = blockBuilder.build();
- if (blocks.isEmpty()) {
- result.add(splitCreator.create(path, 0, length, length,
- targetFileSplitSize, modificationTime, null,
- partitionValues));
- updateCurrentMaxSplitSize();
- return result;
- }
+ return new Iterator<Split>() {
+ private long offset;
+ private int blockIndex;
- long splitStart = start;
- int currentBlockIdx = 0;
- while (splitStart < start + length) {
- updateCurrentMaxSplitSize();
- long splitBytes;
- long remainingBlockBytes = blocks.get(currentBlockIdx).getEnd() -
splitStart;
- if (remainingBlockBytes <= currentMaxSplitSize) {
- splitBytes = remainingBlockBytes;
- } else if (currentMaxSplitSize * 2 >= remainingBlockBytes) {
- // Second to last split in this block, generate two evenly
sized splits
- splitBytes = remainingBlockBytes / 2;
- } else {
- splitBytes = currentMaxSplitSize;
+ @Override
+ public boolean hasNext() {
+ return offset < length;
}
- result.add(splitCreator.create(path, splitStart, splitBytes,
- length, targetFileSplitSize, modificationTime,
blocks.get(currentBlockIdx).getHosts(),
- partitionValues));
- splitStart += splitBytes;
- if (splitStart == blocks.get(currentBlockIdx).getEnd()) {
- currentBlockIdx++;
- if (currentBlockIdx != blocks.size()) {
- Verify.verify(splitStart ==
blocks.get(currentBlockIdx).getStart());
+
+ @Override
+ public Split next() {
+ if (!hasNext()) {
+ throw new NoSuchElementException();
+ }
+ try {
+ long bytes;
+ String[] hosts;
+ if (!canSplit) {
+ bytes = length;
+ hosts = locations.length == 0 ? null :
locations[0].getHosts();
+ nextMaxSplitSize();
+ } else if (specifiedFileSplitSize > 0) {
+ long remaining = length - offset;
+ bytes = (double) remaining / specifiedFileSplitSize >
1.1D
+ ? specifiedFileSplitSize : remaining;
+ int index = getBlockIndex(locations, offset);
+ hosts = index == -1 ? null :
locations[index].getHosts();
+ } else if (blocks.isEmpty()) {
+ bytes = length;
+ hosts = null;
+ nextMaxSplitSize();
+ } else {
+ long maxSize = nextMaxSplitSize();
+ InternalBlock block = blocks.get(blockIndex);
+ long remaining = block.getEnd() - offset;
+ bytes = remaining <= maxSize ? remaining
+ : maxSize * 2 >= remaining ? remaining / 2 :
maxSize;
+ hosts = block.getHosts();
+ }
+ Split result = splitCreator.create(path, offset, bytes,
length, targetFileSplitSize,
+ modificationTime, hosts, partitionValues);
+ offset += bytes;
+ if (canSplit && specifiedFileSplitSize <= 0 &&
!blocks.isEmpty()
+ && offset == blocks.get(blockIndex).getEnd()) {
+ blockIndex++;
+ if (blockIndex < blocks.size()) {
+ Verify.verify(offset ==
blocks.get(blockIndex).getStart());
+ }
+ }
+ return result;
+ } catch (IOException e) {
Review Comment:
This finding does not apply to reviewed head
7e8a562c31a413d44a51decc8588e98ea1935fba. The try body also calls Hadoop
BlockLocation.getHosts() in the unsplittable and explicit-size branches; the
actual Hadoop 3.4.2 dependency declares public String[] getHosts() throws
java.io.IOException (confirmed with javap). That checked exception must be
adapted because Iterator.next cannot declare it. SplitCreator.create need not
throw. The exact-head remote COMPILE build 1089472 succeeded, and local full FE
packaging/Checkstyle also passed. I retained this catch and the lazy
UncheckedIOException/eager IOException boundary. The repair head
f69a30714803c1e2ac476da90b339f0c8838c8f4 again passes all 90 selected FE tests
and full FE packaging/Checkstyle.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -696,6 +696,14 @@ private void initPrunedPartitions() throws UserException {
ensureHmsRuntimeGeneration();
}
+ @Override
+ protected SplitAssignment createSplitAssignment(Map<String, String>
properties,
+ List<String> partitionKeys, boolean admission) {
+ // Hudi still submits eager partition batches and must not block on
Hive's small queues.
+ return new SplitAssignment(backendPolicy, this, this::splitToScanRange,
Review Comment:
Fixed in f69a30714803c1e2ac476da90b339f0c8838c8f4. Hudi now registers a
statement-only cleanup wrapper that captures this query executor and checks its
deferred flag when the statement closes. The actual BatchFsViewOwner remains an
unconditional SplitAssignment resource, so coordinator finalization/query
cancellation still cancels tasks and releases its lease after terminal
accounting. Capturing the executor preserves the handoff even if processor
cleanup clears it from ConnectContext. A regression using real
HudiScanNode.startSplit, StatementContext, SplitAssignment and source
registration reproduced premature stop before repair; it now retains the queued
producer/source through deferred statement close, then verifies explicit scan
stop cancels the producer, removes the source and releases the lease exactly
once. Ordinary and absent-executor statement cleanup remain covered. All 90
selected FE tests and full FE packaging/Checkstyle pass locally; no live
Flight/Hudi external regre
ssion is claimed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
