branch-4.1: [feature](external) support partition- and bucket-aware concurrent writes - #66685
branch-4.1: [feature](external) support partition- and bucket-aware concurrent writes#66685suxiaogang223 wants to merge 13 commits into
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
suxiaogang223
changed the title
2857f25 to
174fc9d
Compare
suxiaogang223
changed the title
|
run buildall |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes for five substantiated issues: one fixed-bucket execution correctness bug, two hard-limit/configuration bugs, and two writer-parallelism bugs.
Review checkpoints:
- Routing and transforms: Paimon BinaryRow/hash parity and Iceberg shared-transform/pinned-spec semantics are sound outside the commented constant-column boundary.
- Planning and scheduling: hidden routing slots survive pruning and translation; SINGLE_WRITER remains one instance in both schedulers. The partitioned legacy cap and partial-statistics issues remain.
- Memory and lifecycle: Arrow/native-page reservation and release ownership, open/abort/close, failed-close retention, and concurrent-writer accounting are balanced. The JVM-limit derivation and live ratio validation are not.
- Variant and compatibility: positional Variant V2 validation, static/reordered inserts, and supported BE-before-FE upgrade sequencing were rechecked without additional findings.
- Tests: the changed tests do not reach the five commented boundaries. Per runner constraints, no builds were run.
- User focus: no additional focus was supplied.
Three complete review rounds converged with no unresolved candidates or duplicate live inline threads.
|
|
||
| bool use_default_implementation_for_nulls() const override { return false; } | ||
|
|
||
| ColumnNumbers get_arguments_that_are_always_constant() const override { |
There was a problem hiding this comment.
[P1] Avoid the all-constant fixed-bucket out-of-bounds path
When all fixed-bucket key expressions are constant on a multi-row input (for example, INSERT ... SELECT 1, payload FROM source), the default constant-argument wrapper retains argument 0 as an N-row ColumnConst but unwraps every key to a one-row nested column. The temporary block therefore still reports N rows, and this loop indexes those one-row key columns at rows 1..N-1. Nullable keys first read past the one-byte null map; fixed/string keys likewise index past their nested column, so a valid insert can crash or misroute rows. Please disable the generic constant fast path here (the implementation already handles ColumnConst) or otherwise make the temporary block cardinalities consistent, and cover multi-row constant and constant-NULL keys.
| public: | ||
| int64_t limit() const { | ||
| const long double limit = | ||
| static_cast<long double>(Jni::Util::get_max_jni_heap_memory_size()) * |
There was a problem hiding this comment.
[P2] Use the real JVM heap on USE_LIBHDFS3 builds
get_max_jni_heap_memory_size() returns SIZE_MAX whenever USE_LIBHDFS3 is defined, so the default ratio makes this expression clamp to INT64_MAX. Those builds still create a finite JVM (-Xmx1g by default, or JAVA_OPTS), and this value is passed to both the process limiter and the Arrow allocator; concurrent Paimon writes therefore have no useful hard cap and can exhaust that JVM. The ordinary parser also misreads valid unitless -Xmx values. Please derive the limit from the created VM/runtime and cover both a finite USE_LIBHDFS3 heap and unitless heap options.
| [](int64_t bytes) -> bool { return bytes > 0; }); | ||
| // Process-wide hard limit as a fraction of the BE JVM -Xmx. | ||
| DEFINE_mDouble(paimon_jni_memory_limit_ratio, "0.5"); | ||
| DEFINE_Validator(paimon_jni_memory_limit_ratio, |
There was a problem hiding this comment.
[P2] Validate the requested mutable ratio
This is a mutable config, but UPDATE_FIELD calls the registered validator before assigning new_value; the generated zero-argument validator therefore reads the old global ratio. Starting from 0.5, set_config(..., "2") (and likewise a negative value or NaN) succeeds and only then stores the invalid value, defeating the (0, 1] hard-limit invariant. Please either make this setting immutable or validate the proposed value before committing it, with a live-update regression case.
| if (connectContext != null && connectContext.getSessionVariable() != null) { | ||
| expectInstanceNum = connectContext.getSessionVariable().getExchangeInstanceParallel(); | ||
| } | ||
| int writerInstanceLimit = fragment.getSink() == null |
There was a problem hiding this comment.
[P2] Enforce writer caps in the legacy scheduler too
This is the only scheduling consumer of getWriterInstanceLimit(). When enable_nereids_distribute_planner=false, the legacy Coordinator instead copies every instance of the largest child fragment (or uses exchange_instance_parallel) for partitioned exchange-root sinks without consulting this cap. A four-bucket Paimon plan that reports planned writers: 4 can therefore still open, for example, 64 JNI writers, defeating the resource/small-writer control on a supported session path. Please apply the same upper bound in the legacy partitioned branch and cover that scheduler mode.
| boolean foundKnownInput = false; | ||
| for (Slot input : expression.getInputSlots()) { | ||
| ColumnStatistic columnStatistic = statistics.findColumnStatistics(input); | ||
| if (columnStatistic == null || columnStatistic.isUnKnown |
There was a problem hiding this comment.
[P2] Keep capacity when any route input is unknown
For a composite routing expression, this skips unknown input statistics but returns the product of the remaining known NDVs as soon as one is available. Thus a fixed-bucket route with an NDV-1 key plus an unknown high-cardinality key is estimated as one ownership unit and capped to one writer, even though the unknown key can fill every bucket. That silently defeats this PR's scaling goal on common incomplete statistics. Please treat the expression cardinality as unknown when any relevant input is unknown (then retain the connector cap/capacity), and add a mixed-known/unknown test.
FE UT Coverage ReportIncrement line coverage |
2 participants
What
This change introduces a connector-independent write-distribution framework for external table sinks, with Iceberg and Paimon as its first consumers.
For Iceberg:
For Paimon:
HASH_FIXEDtables using Paimon-compatible native routing(partition, bucket)to one writer within a write jobHASH_DYNAMICKEY_DYNAMICThe change also bounds Paimon JNI writer memory:
JVM -Xmx * paimon_jni_memory_limit_ratio0.5as the default ratioWhy
External table writers previously lacked a common way to express the ownership unit that should be used for writer distribution.
For Iceberg, routing by raw source columns does not reproduce transforms such as
bucket,truncate, or time transforms. Rows belonging to the same physical Iceberg partition could therefore be distributed inconsistently, reducing the effectiveness of concurrent writing and producing unnecessary small files.For Paimon fixed-bucket tables, concurrent writing must preserve
(partition, bucket)ownership. Sending the same bucket to multiple writers may create conflicting writer state and additional files. The previous safe fallback serialized more Paimon writes than necessary.This change separates connector-specific ownership calculation from the common Doris scheduling path:
This provides a reusable foundation for other external table formats without adding format-specific exchange protocols.
Scope
This change only manages writer distribution and writer parallelism. It does not change connector file-rolling policies or target file-size settings.
Stateful Paimon dynamic-bucket assignment and global-index assignment are intentionally not implemented in this change. Those modes continue to use the existing single-writer fallback.