[fix](be) Clarify Hive and Iceberg partition writer limit errors

be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp

@@ -570,8 +570,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));

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));

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

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)"""
+    }
+}

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)"""
+    }
+}