github-actions[bot] commented on code in PR #66802:
URL: https://github.com/apache/doris/pull/66802#discussion_r3822031891
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -571,6 +572,69 @@ private TScanRangeLocations splitToScanRange(
return curLocations;
}
+ static void setTargetSplitSize(TFileRangeDesc rangeDesc, SessionVariable
sessionVariable) {
+ setTargetSplitSize(rangeDesc, sessionVariable, true);
+ }
+
+ private static void setTargetSplitSize(
+ TFileRangeDesc rangeDesc, SessionVariable sessionVariable, boolean
supportsBeSplit) {
+ if (supportsBeSplit && canSplitOnBe(rangeDesc, sessionVariable)) {
+
rangeDesc.setTargetSplitSize(sessionVariable.getFileSplitSizeOnBe());
+ }
+ }
+
+ protected long selectFeSplitSizeForBe(
+ long fallbackSize, TFileFormatType format, boolean
supportsBeSplit) {
+ // FE may enlarge a source range only when BE can refine it. Otherwise
the larger range
+ // would silently reduce the parallelism provided by the legacy split
planner.
+ if (!supportsBeSplit || !isBeSplitEnabled(sessionVariable) ||
wouldRunSerialOnBe()
+ || !isColumnarFormat(format)) {
+ return fallbackSize;
+ }
+ return sessionVariable.getFileSplitSizeOnFe();
Review Comment:
[P2] Preserve `max_file_split_num` after selecting the coarse target. The
changed connectors raise their legacy fallback with
`applyMaxFileSplitNumLimit()` before this helper, but an eligible range then
discards that result and always returns `file_split_size_on_fe`. For example, 2
GiB with `max_file_split_num=1` computes a 2 GiB cap-derived target and is
changed back to 512 MiB, creating four FE ranges despite the configured bound.
Apply the cap to the final eligible target (or carry its lower bound
separately, without making explicit legacy `file_split_size` override the
intended coarse target) and add capped connector tests.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java:
##########
@@ -210,23 +210,27 @@ private List<Split> getLanceSplits() throws UserException
{
return splits;
}
- private long determineTargetFileSplitSize(List<TBrokerFileStatus>
fileStatuses) {
+ private long determineTargetFileSplitSize(List<TBrokerFileStatus>
fileStatuses) throws UserException {
+ long fallbackSize;
if (sessionVariable.getFileSplitSize() > 0) {
- return sessionVariable.getFileSplitSize();
- }
- long result = sessionVariable.getMaxInitialSplitSize();
- long totalFileSize = 0;
- boolean exceedInitialThreshold = false;
- for (TBrokerFileStatus fileStatus : fileStatuses) {
- totalFileSize += fileStatus.getSize();
- if (!exceedInitialThreshold
- && totalFileSize >= sessionVariable.getMaxSplitSize() *
sessionVariable.getMaxInitialSplitNum()) {
- exceedInitialThreshold = true;
+ fallbackSize = sessionVariable.getFileSplitSize();
+ } else {
+ long totalFileSize = 0;
+ boolean exceedInitialThreshold = false;
+ for (TBrokerFileStatus fileStatus : fileStatuses) {
+ totalFileSize += fileStatus.getSize();
+ if (!exceedInitialThreshold
+ && totalFileSize
+ >= sessionVariable.getMaxSplitSize() *
sessionVariable.getMaxInitialSplitNum()) {
+ exceedInitialThreshold = true;
+ }
}
+ fallbackSize = exceedInitialThreshold
+ ? sessionVariable.getMaxSplitSize() :
sessionVariable.getMaxInitialSplitSize();
+ fallbackSize = applyMaxFileSplitNumLimit(fallbackSize,
totalFileSize);
}
- result = exceedInitialThreshold ? sessionVariable.getMaxSplitSize() :
result;
- result = applyMaxFileSplitNumLimit(result, totalFileSize);
- return result;
+ return selectFeSplitSizeForBe(
Review Comment:
[P2] Keep zero-version HTTP files on the legacy FE split size. HTTP TVFs
create splittable statuses without a modification time, so this returns the 512
MiB coarse target, but both Parquet and ORC later refuse
`build_physical_splits()` when mtime is zero and the object is not immutable. A
file that previously had multiple 32/64 MiB ranges can therefore become one
range consumed by one scanner, while scanner construction is still uncapped.
Include stable identity in eligibility (or carry a version/ETag contract) and
cover zero-mtime HTTP Parquet and ORC.
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -671,6 +671,8 @@ struct TFileRangeDesc {
// whether the value of columns_from_path is null
15: optional list<bool> columns_from_path_is_null;
16: optional bool file_cache_admission;
+ // FE's effective split target in bytes for BE-local physical split
refinement.
+ 17: optional i64 target_split_size;
Review Comment:
[P2] Treat this field as an explicit refinement handshake. With a
nonpositive dedicated size, current FE retains legacy ranges and omits field
17, but current BE still uncaps/refines them and without a target can create
one task per small Row Group/Stripe; metadata-COUNT ranges likewise omit the
field yet over-allocate scanners. Old-FE/new-BE has the same absent-field
behavior, while new-FE/old-BE sends coarse ranges to a BE that ignores the
field and cannot refine. Require field presence before BE scanner
uncap/refinement and gate coarse FE sizing on backend capability; cover
disabled sizes, metadata COUNT, and both upgrade directions.
##########
be/src/format_v2/orc/orc_reader.h:
##########
@@ -49,12 +65,15 @@ class OrcReader final : public format::FileReader {
std::unique_ptr<io::FileDescription>& file_description,
std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile,
std::optional<format::GlobalRowIdContext> global_rowid_context =
std::nullopt,
- bool enable_mapping_timestamp_tz = false);
+ bool enable_mapping_timestamp_tz = false,
+ std::shared_ptr<const FileContext> file_context = nullptr);
~OrcReader() override;
static format::ColumnDefinition row_position_column_definition();
Status init(RuntimeState* state) override;
+ Status build_physical_splits(std::vector<PhysicalFileSplit>* splits,
Review Comment:
[P2] Expose ORC's metadata aggregate through the planning-reader hook.
`TableReader` now retains the source reader only when
`get_metadata_aggregate_result()` succeeds, but ORC inherits the default
`NotSupported` even though its `get_aggregate_result()` computes COUNT and
MIN/MAX solely from selected-stripe metadata. Multi-stripe aggregates therefore
close the ready reader and repeat reader/aggregate lifecycles per child (and
MIN/MAX emits 2C synthetic rows for C children). Override the metadata hook and
retain the source on success, with child fallback on `NotSupported`.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileQueryScanNode.java:
##########
@@ -571,6 +572,69 @@ private TScanRangeLocations splitToScanRange(
return curLocations;
}
+ static void setTargetSplitSize(TFileRangeDesc rangeDesc, SessionVariable
sessionVariable) {
+ setTargetSplitSize(rangeDesc, sessionVariable, true);
+ }
+
+ private static void setTargetSplitSize(
+ TFileRangeDesc rangeDesc, SessionVariable sessionVariable, boolean
supportsBeSplit) {
+ if (supportsBeSplit && canSplitOnBe(rangeDesc, sessionVariable)) {
+
rangeDesc.setTargetSplitSize(sessionVariable.getFileSplitSizeOnBe());
+ }
+ }
+
+ protected long selectFeSplitSizeForBe(
+ long fallbackSize, TFileFormatType format, boolean
supportsBeSplit) {
+ // FE may enlarge a source range only when BE can refine it. Otherwise
the larger range
+ // would silently reduce the parallelism provided by the legacy split
planner.
+ if (!supportsBeSplit || !isBeSplitEnabled(sessionVariable) ||
wouldRunSerialOnBe()
+ || !isColumnarFormat(format)) {
+ return fallbackSize;
+ }
+ return sessionVariable.getFileSplitSizeOnFe();
+ }
+
+ private boolean wouldRunSerialOnBe() {
+ // Mirror the external-scan branch of
ScanLocalState::should_run_serial so FE does not
+ // enlarge ranges when BE intentionally constructs only one scanner
for a small LIMIT.
+ return sessionVariable.enableAdaptivePipelineTaskSerialReadOnLimit
+ && conjuncts.isEmpty()
+ && getLimit() > 0
+ && getLimit() <=
sessionVariable.adaptivePipelineTaskSerialReadOnLimit;
+ }
+
+ static boolean canSplitOnBe(TFileRangeDesc rangeDesc, SessionVariable
sessionVariable) {
+ if (!isBeSplitEnabled(sessionVariable) || !rangeDesc.isSetFormatType()
+ || !isColumnarFormat(rangeDesc.getFormatType())) {
+ return false;
+ }
+ if (!rangeDesc.isSetTableFormatParams()) {
+ return true;
+ }
+ if (rangeDesc.getTableFormatParams().isSetTableLevelRowCount()
+ && rangeDesc.getTableFormatParams().getTableLevelRowCount() >=
0) {
+ return false;
+ }
+ if (TableFormatType.TRANSACTIONAL_HIVE.value().equals(
+ rangeDesc.getTableFormatParams().getTableFormatType())) {
+ return false;
+ }
+ return !rangeDesc.getTableFormatParams().isSetIcebergParams()
+ ||
!rangeDesc.getTableFormatParams().getIcebergParams().isSetDeleteFiles()
+ ||
rangeDesc.getTableFormatParams().getIcebergParams().getDeleteFiles().isEmpty();
+ }
+
+ private static boolean isBeSplitEnabled(SessionVariable sessionVariable) {
+ return sessionVariable.enableFileScannerV2
+ && sessionVariable.maxFileScannersConcurrency != 1
Review Comment:
[P2] Match coarse eligibility to the actual BE consumer count, not only the
session cap. A lone range is safely serialized, but non-serial skew remains:
with one BE, effective scanner budget P, P coarse ranges, and one 400 MiB
source plus P-1 tiny sources, FE keeps P operator instances; BE constructs `P /
P = 1` scanner per local state and then refuses refinement. The large source
becomes a single-scanner tail, whereas legacy 64 MiB ranges distributed it
across instances. The existing BE guard only avoids child-lifecycle overhead;
add a capability/count contract or preserve FE distribution when fewer than two
consumers are guaranteed.
--
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]
