github-actions[bot] commented on code in PR #69986:
URL: https://github.com/apache/doris/pull/69986#discussion_r4449297781
##########
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:
[P1] Remove the unreachable checked IOException catch in splitFileIterator's
anonymous Iterator.next(). SplitCreator.create(...) has no throws IOException,
and every other call in that try body is non-throwing, so javac reports
`exception IOException is never thrown in body of corresponding try statement`
and the FE module cannot compile. Either remove the catch/wrapper or change the
SplitCreator API and all implementations to a checked contract.
##########
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:
[P1] Keep deferred Arrow Flight Hudi split owners alive until coordinator
finalization. FileQueryScanNode now dispatches batch scans through this
Hudi-specific assignment factory, but Hudi registers BatchFsViewOwner as a
StatementContext resource and its close() unconditionally calls
splitAssignment.stop(). StatementContext closes resources at the end of
GetFlightInfo, so a Hudi batch scan is stopped before the BE's later
DoGet/fetchSplitBatch, even though FileQueryScanNode explicitly preserves
deferred scans. Make this owner cleanup defer-aware while retaining
cancellation/connection teardown, and add a Hudi deferred-batch regression.
--
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]
