github-actions[bot] commented on code in PR #68837:
URL: https://github.com/apache/doris/pull/68837#discussion_r4244166697
##########
gensrc/thrift/Partitions.thrift:
##########
@@ -45,20 +45,21 @@ enum TPartitionType {
// used for shuffle data by parititon and tablet
OLAP_TABLE_SINK_HASH_PARTITIONED = 6,
- // used for shuffle data by hive parititon
- HIVE_TABLE_SINK_HASH_PARTITIONED = 7,
+ // used for shuffle data by external table sink ownership key
+ EXTERNAL_TABLE_SINK_HASH_PARTITIONED = 7,
- // used for hive unparititoned table
- HIVE_TABLE_SINK_UNPARTITIONED = 8,
+ // adaptive writer distribution without an ownership key
+ EXTERNAL_TABLE_SINK_UNPARTITIONED = 8,
// used for merge partitioning: insert by partition columns, delete by row_id
MERGE_PARTITIONED = 9,
- // connector-owned ownership function followed by writer assignment
- EXTERNAL_TABLE_SINK_HASH_PARTITIONED = 10,
+ // used for shuffle data by hive parititon. These master-only values were
moved
+ // before release so values 7 and 8 keep the branch-4.1 external-sink
meaning.
+ HIVE_TABLE_SINK_HASH_PARTITIONED = 10,
Review Comment:
[P1] Preserve Hive routing when an older BE can execute the fragment. With
`be_exec_version=13` for a rolling upgrade, the new FE emits value 10 for
partitioned Hive writes (11 for unpartitioned parallel writes), but 4.1/4.2 BEs
know only values through 9. Their release-mode exchange sink has no handler for
either value; its fallback sends no block and returns OK, so an INSERT can
commit without its rows. Gate this plan on BE capability or provide compatible
routing, and fail unknown kinds.
##########
gensrc/thrift/Partitions.thrift:
##########
@@ -205,12 +206,35 @@ enum TExternalTableSinkWriterAssignment {
SKEWED = 1
}
-// FE treats partition_function and its options as opaque connector-owned data.
-// BE validates the named function before processing rows.
+// Deprecated branch-4.1 routing algorithm. Kept at its original field id in
+// TExternalTableSinkHashPartitionInfo for wire compatibility.
+enum TExternalTableSinkHashAlgorithm {
+ DIRECT_HASH = 0,
+ ICEBERG_TRANSFORM = 1,
+ PAIMON_FIXED_BUCKET = 2
+}
+
+// Deprecated branch-4.1 Paimon routing metadata. The generic connector-owned
+// partition function encodes this information in partition_function_options.
+struct TPaimonFixedBucketInfo {
+ 1: required i32 num_buckets
+ 2: required list<i32> partition_field_indexes
+ 3: required list<i32> bucket_field_indexes
+}
+
+// Fields 1-4 retain their branch-4.1 meanings and are deprecated on master.
+// Master appends its generic connector-owned function representation instead
+// of reusing an existing field id for a different type or meaning.
struct TExternalTableSinkHashPartitionInfo {
- 1: required string partition_function
- 2: optional map<string, string> partition_function_options
+ 1: optional TExternalTableSinkHashAlgorithm algorithm
+ 2: optional list<string> partition_transforms
3: required TExternalTableSinkWriterAssignment writer_assignment
+ 4: optional TPaimonFixedBucketInfo paimon_fixed_bucket_info
+
+ // FE treats these fields as opaque connector-owned data. BE validates the
+ // named function before processing rows.
+ 5: optional string partition_function
Review Comment:
[P1] Handle both external hash routing encodings during rolling upgrades.
With `be_exec_version=13`, the new FE fills only fields 3/5/6 while 4.1/4.2 BEs
require algorithm field 1; the reverse pair sends 1/3/2/4, but this BE selects
solely from field 5 and rejects the empty function. The existing version gate
permits these writes, so fixed-bucket Paimon inserts fail in either
mixed-version direction. Emit and consume a compatible dialect, or reject
mixed-version plans before dispatch.
##########
gensrc/thrift/FrontendService.thrift:
##########
@@ -339,8 +339,8 @@ struct TReportExecStatusParams {
33: optional string first_error_msg
- // Opaque, connector-owned commit fragments; FE routes them to the
transaction.
- 34: optional list<binary> connector_commit_data
+ // Paimon commit messages prepared by BE and committed by FE.
+ 34: optional list<DataSinks.TPaimonCommitMessage> paimon_commit_messages
Review Comment:
[P1] Keep field 34 decodable across rolling master upgrades. The pre-PR
master sends serialized `TPaimonCommitMessage` blobs as `list<binary>`, while
this change reuses id 34 for `list<TPaimonCommitMessage>` (STRING versus STRUCT
elements). A new BE's report cannot be decoded by an older master FE; the
reverse pair also loses the payload when its binary length is read as struct
data. Both generations allow Paimon writes at execution version 16 and already
have the ACK field, so this is separate from the 4.1/4.2 missing-ACK issue. Use
a new optional typed field with negotiation, or block mixed-master writes
before preparing files.
##########
be/src/exec/pipeline/pipeline_fragment_context.cpp:
##########
@@ -2602,7 +2602,7 @@ void PipelineFragmentContext::_coordinator_callback(const
ReportStatusRequest& r
}
const bool requires_external_file_ack =
- params.__isset.iceberg_commit_datas ||
params.__isset.connector_commit_data;
+ params.__isset.iceberg_commit_datas ||
params.__isset.paimon_commit_messages;
Review Comment:
[P1] Account for older FE responses before requiring this Paimon ACK.
Branch-4.1/4.2 FEs already consume typed field-34 `paimon_commit_messages`, but
their `TReportExecStatusResult` has only `status`; a master BE therefore treats
every nonempty Paimon report to those FEs as rejected and cancels the write
after FE may have accepted it. Use an explicit compatible acceptance path or
block this mixed-version write before metadata transfer; an absent new field
cannot prove rejection by an older FE.
##########
be/src/runtime/runtime_state.cpp:
##########
@@ -84,22 +84,32 @@ Status
RuntimeState::add_iceberg_commit_datas(TIcebergCommitData iceberg_commit_
return Status::OK();
}
-Status RuntimeState::add_connector_commit_data(std::string commit_data) {
+Status
RuntimeState::add_paimon_commit_messages(std::vector<TPaimonCommitMessage>
commit_messages) {
+ size_t serialized_bytes = 0;
+ ThriftSerializer serializer(false, 256);
+ for (auto& commit_message : commit_messages) {
+ uint32_t serialized_size = 0;
+ uint8_t* buffer = nullptr;
+ RETURN_IF_ERROR(serializer.serialize(&commit_message,
&serialized_size, &buffer));
+ serialized_bytes += serialized_size + sizeof(uint32_t);
+ }
+
constexpr size_t report_envelope_headroom = 1024 * 1024;
const size_t thrift_limit = coordinator_thrift_message_limit();
const size_t commit_data_limit =
thrift_limit > report_envelope_headroom ? thrift_limit -
report_envelope_headroom : 0;
std::lock_guard<std::mutex>
budget_lock(_external_file_report_state->mutex);
- if (_external_file_report_state->serialized_commit_bytes +
commit_data.size() +
- sizeof(uint32_t) >
+ if (_external_file_report_state->serialized_commit_bytes +
serialized_bytes >
Review Comment:
[P2] Preserve an abort path when the whole Paimon batch exceeds the report
budget. `PaimonCommitCodec` can produce multiple chunks; this new sum check
rejects all of them after `PaimonTableWriter::close` has already closed the
Java writer and cleared its prepared messages. The final error report then
carries no payloads for this writer, so FE rollback cannot call Paimon abort
for its files. Previously, if the first chunk fit and a later one overflowed,
that prefix still reached FE rollback. Reserve the budget while the writer can
still abort, or retain a safe cleanup handle before rejecting the batch.
##########
gensrc/thrift/PlanNodes.thrift:
##########
@@ -395,9 +396,13 @@ struct TPaimonFileDesc {
16: optional i64 schema_id; // for schema change.
// Reader implementation for logical paimon split. Native file split uses
range format type.
17: optional TPaimonReaderType reader_type;
+ // Reserved for the Paimon Rust reader. Keep these ids aligned with
branch-4.1.
+ 18: optional string paimon_table_schema_json
+ 19: optional string paimon_branch
// Original Paimon RawFile.path() before Doris storage path normalization.
Native readers use this
- // to materialize the public file-location metadata column.
- 18: optional string original_file_path;
+ // to materialize the public file-location metadata column. This
master-only field is appended after
+ // the branch-4.1 fields instead of reusing field 18.
+ 20: optional string original_file_path;
Review Comment:
[P2] Preserve the original path for a pre-PR master BE during a rolling
upgrade. That BE already supports Paimon
`__paimon_file_path`/`__paimon_row_index` but reads `original_file_path` from
field 18; the new FE sends only field 20, so it receives an empty path and
fails those scans. The planner's `isSmoothUpgradeSrc` check covers Cloud
replacement BEs, not ordinary rolling BEs, and both master versions use
execution version 16. Keep this path decodable across the mixed-master pair or
gate these scans on a per-BE capability.
--
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]
