This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f44fd38a762 [fix](be) Clarify Hive and Iceberg partition writer limit
errors (#69183)
f44fd38a762 is described below
commit f44fd38a762700c03fecea31b5045857da1c67fb
Author: Socrates <[email protected]>
AuthorDate: Thu Oct 22 11:12:49 2026 +0800
[fix](be) Clarify Hive and Iceberg partition writer limit errors (#69183)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: A Hive or Iceberg writer that tries to exceed its
configured open-partition limit currently returns `INTERNAL_ERROR: Too
many open partitions 128`, without explaining the scope or how to
proceed.
Return `INVALID_ARGUMENT` and include the writer type, current and
attempted partition counts, limit, BE configuration
`table_sink_partition_write_max_partition_nums_per_writer`, and advice
to split the write or increase the configuration. The existing
per-writer limit check is unchanged.
Add BE tests for both writer entry points at default and configured
limits, plus Hive and Iceberg SQL suites covering writes at the limit,
rejection above the limit, and success after increasing it.
### Release note
Hive and Iceberg writes exceeding the per-writer open-partition limit
now report `INVALID_ARGUMENT` with the limit and actionable
configuration guidance.
### Check List (For Author)
- Test: Formatting, build hygiene and Groovy syntax checks passed. Local
BE UT was blocked by missing `OpenMP_C` during CMake configuration;
clang-tidy needs the unavailable compilation database. External
regression suites were not run because Docker is unavailable.
- Behavior changed: Yes, error classification and message only.
- Does this need documentation: No.
---
.../sink/writer/iceberg/viceberg_table_writer.cpp | 10 +-
be/src/exec/sink/writer/vhive_table_writer.cpp | 10 +-
.../external_table_writer_partition_limit_test.cpp | 111 +++++++++++++++++++++
.../write/test_hive_write_partition_limit.groovy | 54 ++++++++++
.../test_iceberg_write_partition_limit.groovy | 60 +++++++++++
5 files changed, 241 insertions(+), 4 deletions(-)
diff --git a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
index bdfed84ee5d..7aa81a7387b 100644
--- a/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
+++ b/be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
@@ -572,8 +572,14 @@ Status VIcebergTableWriter::_write_prepared_block(Block&
output_block) {
if (writer_iter == _partitions_to_writers.end()) {
if (_partitions_to_writers.size() + 1 >
config::table_sink_partition_write_max_partition_nums_per_writer) {
- return Status::InternalError(
- "Too many open partitions {}",
+ return Status::InvalidArgument(
+ "Too many open partitions for Iceberg writer: open
partitions={}, "
+ "attempted partitions={}, limit={}. This is a
per-writer limit "
+ "controlled by BE configuration "
+
"'table_sink_partition_write_max_partition_nums_per_writer'. "
+ "Split the write into multiple statements or
increase this "
+ "configuration.",
+ _partitions_to_writers.size(),
_partitions_to_writers.size() + 1,
config::table_sink_partition_write_max_partition_nums_per_writer);
}
RETURN_IF_ERROR(create_and_open_writer(partition_name, i,
nullptr, 0, writer));
diff --git a/be/src/exec/sink/writer/vhive_table_writer.cpp
b/be/src/exec/sink/writer/vhive_table_writer.cpp
index 04a0c04c1dd..5d68d50caa3 100644
--- a/be/src/exec/sink/writer/vhive_table_writer.cpp
+++ b/be/src/exec/sink/writer/vhive_table_writer.cpp
@@ -176,8 +176,14 @@ Status VHiveTableWriter::write(RuntimeState* state, Block&
block) {
std::shared_ptr<VHivePartitionWriter> writer;
if (_partitions_to_writers.size() + 1 >
config::table_sink_partition_write_max_partition_nums_per_writer) {
- return Status::InternalError(
- "Too many open partitions {}",
+ return Status::InvalidArgument(
+ "Too many open partitions for Hive writer: open
partitions={}, "
+ "attempted partitions={}, limit={}. This is a
per-writer limit "
+ "controlled by BE configuration "
+
"'table_sink_partition_write_max_partition_nums_per_writer'. "
+ "Split the write into multiple statements or
increase this "
+ "configuration.",
+ _partitions_to_writers.size(),
_partitions_to_writers.size() + 1,
config::table_sink_partition_write_max_partition_nums_per_writer);
}
RETURN_IF_ERROR(create_and_open_writer(partition_name, i,
nullptr, 0, writer));
diff --git
a/be/test/exec/sink/writer/external_table_writer_partition_limit_test.cpp
b/be/test/exec/sink/writer/external_table_writer_partition_limit_test.cpp
new file mode 100644
index 00000000000..6e611654cdc
--- /dev/null
+++ b/be/test/exec/sink/writer/external_table_writer_partition_limit_test.cpp
@@ -0,0 +1,111 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include "common/config.h"
+#include "core/block/block.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "exec/sink/writer/iceberg/viceberg_table_writer.h"
+#include "exec/sink/writer/vhive_table_writer.h"
+#include "exprs/vexpr_context.h"
+#include "exprs/vslot_ref.h"
+#include "runtime/runtime_state.h"
+
+namespace doris {
+
+class ExternalTableWriterPartitionLimitTest : public
testing::TestWithParam<int32_t> {
+protected:
+ void SetUp() override {
+ _original_limit =
config::table_sink_partition_write_max_partition_nums_per_writer;
+ config::table_sink_partition_write_max_partition_nums_per_writer =
GetParam();
+ }
+
+ void TearDown() override {
+ config::table_sink_partition_write_max_partition_nums_per_writer =
_original_limit;
+ }
+
+ Block new_partition_block() {
+ auto column = ColumnInt32::create();
+ column->get_data().push_back(GetParam());
+ return Block(
+ {ColumnWithTypeAndName(std::move(column),
std::make_shared<DataTypeInt32>(), "p")});
+ }
+
+ void check_limit_status(const Status& status, const std::string&
writer_type) {
+ EXPECT_EQ(TStatusCode::INVALID_ARGUMENT,
status.to_thrift().status_code);
+ const auto message = status.to_string();
+ EXPECT_NE(std::string::npos, message.find(writer_type + " writer"));
+ EXPECT_NE(std::string::npos, message.find("open partitions=" +
std::to_string(GetParam())));
+ EXPECT_NE(std::string::npos,
+ message.find("attempted partitions=" +
std::to_string(GetParam() + 1)));
+ EXPECT_NE(std::string::npos, message.find("limit=" +
std::to_string(GetParam())));
+ EXPECT_NE(std::string::npos, message.find("per-writer limit"));
+ EXPECT_NE(std::string::npos,
+
message.find("table_sink_partition_write_max_partition_nums_per_writer"));
+ EXPECT_NE(std::string::npos, message.find("Split the write into
multiple statements"));
+ }
+
+private:
+ int32_t _original_limit = 0;
+};
+
+TEST_P(ExternalTableWriterPartitionLimitTest,
HiveRejectsNextPartitionBeforeOpeningAFile) {
+ THiveColumn partition_column;
+ partition_column.__set_name("p");
+ partition_column.__set_column_type(THiveColumnType::PARTITION_KEY);
+ THiveTableSink hive_sink;
+ hive_sink.__set_columns({partition_column});
+ TDataSink sink;
+ sink.__set_hive_table_sink(hive_sink);
+ auto type = std::make_shared<DataTypeInt32>();
+ VExprContextSPtrs exprs = {
+ VExprContext::create_shared(VSlotRef::create_shared(0, 0, -1,
type, "p"))};
+ VHiveTableWriter writer(sink, exprs, nullptr, nullptr);
+ writer._partition_columns_input_index = {0};
+ // Existing writers are never dereferenced when rejecting a new partition.
No filesystem is needed.
+ for (int32_t i = 0; i < GetParam(); ++i) {
+ writer._partitions_to_writers.emplace("p=" + std::to_string(i),
nullptr);
+ }
+ RuntimeState state;
+ auto block = new_partition_block();
+ check_limit_status(writer.write(&state, block), "Hive");
+ EXPECT_EQ(static_cast<size_t>(GetParam()),
writer._partitions_to_writers.size());
+}
+
+TEST_P(ExternalTableWriterPartitionLimitTest,
IcebergRejectsNextPartitionBeforeOpeningAFile) {
+ TDataSink sink;
+ sink.__set_iceberg_table_sink(TIcebergTableSink());
+ VExprContextSPtrs exprs;
+ iceberg::PartitionField field(1, 1000, "p", "identity");
+ VIcebergTableWriter writer(sink, exprs, nullptr, nullptr);
+ writer._iceberg_partition_columns.emplace_back(
+ field, TYPE_INT, 0, std::vector<size_t> {},
+
std::make_unique<IdentityPartitionColumnTransform>(std::make_shared<DataTypeInt32>()));
+ for (int32_t i = 0; i < GetParam(); ++i) {
+ writer._partitions_to_writers.emplace("existing-" + std::to_string(i),
nullptr);
+ }
+ auto block = new_partition_block();
+ check_limit_status(writer.write_prepared_block(block), "Iceberg");
+ EXPECT_EQ(static_cast<size_t>(GetParam()),
writer._partitions_to_writers.size());
+}
+
+INSTANTIATE_TEST_SUITE_P(DefaultAndConfiguredLimits,
ExternalTableWriterPartitionLimitTest,
+ testing::Values(128, 2));
+
+} // namespace doris
diff --git
a/regression-test/suites/external_table_p0/hive/write/test_hive_write_partition_limit.groovy
b/regression-test/suites/external_table_p0/hive/write/test_hive_write_partition_limit.groovy
new file mode 100644
index 00000000000..960f9144e36
--- /dev/null
+++
b/regression-test/suites/external_table_p0/hive/write/test_hive_write_partition_limit.groovy
@@ -0,0 +1,54 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_hive_write_partition_limit",
+ "p0,external,hive,external_docker,external_docker_hive,nonConcurrent")
{
+ String enabled = context.config.otherConfigs.get("enableHiveTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable hive test")
+ return
+ }
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ sql """drop catalog if exists test_hive_write_partition_limit"""
+ setHivePrefix("hive2")
+ String hmsPort = context.config.otherConfigs.get("hive2HmsPort")
+ sql """create catalog test_hive_write_partition_limit properties (
+ "type" = "hms",
+ "hive.metastore.uris" = "thrift://${externalEnvIp}:${hmsPort}",
+ "hadoop.username" = "hive"
+ )"""
+ hive_docker """drop table if exists hive_write_partition_limit"""
+ hive_docker """create table hive_write_partition_limit (id int)
+ partitioned by (p int) stored as parquet"""
+ sql """use test_hive_write_partition_limit.`default`"""
+
+ // The limit belongs to each writer, so route all VALUES rows through one
writer.
+ sql """set parallel_pipeline_task_num = 1"""
+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer":
"2"]) {
+ // Exactly two partitions are allowed; the next partition must fail
with actionable context.
+ sql """insert into hive_write_partition_limit values (0, 0), (1, 1)"""
+ test {
+ sql """insert into hive_write_partition_limit values (2, 2), (3,
3), (4, 4)"""
+ exception "[INVALID_ARGUMENT]Too many open partitions for Hive
writer: " +
+ "open partitions=2, attempted partitions=3, limit=2."
+ }
+ }
+ // The advertised configuration is effective: the same three-partition
write can now succeed.
+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer":
"3"]) {
+ sql """insert into hive_write_partition_limit values (2, 2), (3, 3),
(4, 4)"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_limit.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_limit.groovy
new file mode 100644
index 00000000000..e3ff3332c12
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_limit.groovy
@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_iceberg_write_partition_limit",
+
"p0,external,iceberg,external_docker,external_docker_iceberg,nonConcurrent") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
+ return
+ }
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ sql """drop catalog if exists test_iceberg_write_partition_limit"""
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ sql """create catalog test_iceberg_write_partition_limit properties (
+ "type" = "iceberg",
+ "iceberg.catalog.type" = "rest",
+ "uri" = "http://${externalEnvIp}:${restPort}";,
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minioPort}";,
+ "s3.region" = "us-east-1"
+ )"""
+ sql """create database if not exists
test_iceberg_write_partition_limit.partition_limit_db"""
+ sql """use test_iceberg_write_partition_limit.partition_limit_db"""
+ sql """drop table if exists iceberg_write_partition_limit"""
+ sql """create table iceberg_write_partition_limit (id int, p int)
+ partition by list (p) ()
+ properties ("format-version" = "2", "write.format.default" =
"parquet")"""
+
+ // The limit belongs to each writer, so route all VALUES rows through one
writer.
+ sql """set parallel_pipeline_task_num = 1"""
+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer":
"2"]) {
+ // Exactly two partitions are allowed; the next partition must fail
with actionable context.
+ sql """insert into iceberg_write_partition_limit values (0, 0), (1,
1)"""
+ test {
+ sql """insert into iceberg_write_partition_limit values (2, 2),
(3, 3), (4, 4)"""
+ exception "[INVALID_ARGUMENT]Too many open partitions for Iceberg
writer: " +
+ "open partitions=2, attempted partitions=3, limit=2."
+ }
+ }
+ // The advertised configuration is effective: the same three-partition
write can now succeed.
+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer":
"3"]) {
+ sql """insert into iceberg_write_partition_limit values (2, 2), (3,
3), (4, 4)"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
