Skip to content

Commit a58cdcf

Browse files
suxiaogang223suxiaogang223
committed
[fix](be) Clarify Hive and Iceberg partition writer limit errors
### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Hive and Iceberg table writers report INTERNAL_ERROR when opening another partition would exceed the configured per-writer partition limit. Use INVALID_ARGUMENT and report the writer type, current and attempted partition counts, the limit, its BE configuration name, and how to split or configure the write. Keep the existing limit check unchanged. Add BE tests for both writer entry points and SQL regression coverage for the boundary, limit violation, and configuration adjustment. ### 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, Groovy syntax and git diff --check passed. BE UT added but local execution was blocked by missing OpenMP_C during CMake configuration. External regression tests added but not run because Docker is unavailable. clang-tidy could not run without a compilation database. - Behavior changed: Yes, error classification and message only; partition limits and write behavior are unchanged. - Does this need documentation: No
1 parent 28c9f8d commit a58cdcf

5 files changed

Lines changed: 241 additions & 4 deletions

File tree

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

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -570,8 +570,14 @@ Status VIcebergTableWriter::_write_prepared_block(Block& output_block) {
570570
if (writer_iter == _partitions_to_writers.end()) {
571571
if (_partitions_to_writers.size() + 1 >
572572
config::table_sink_partition_write_max_partition_nums_per_writer) {
573-
return Status::InternalError(
574-
"Too many open partitions {}",
573+
return Status::InvalidArgument(
574+
"Too many open partitions for Iceberg writer: open partitions={}, "
575+
"attempted partitions={}, limit={}. This is a per-writer limit "
576+
"controlled by BE configuration "
577+
"'table_sink_partition_write_max_partition_nums_per_writer'. "
578+
"Split the write into multiple statements or increase this "
579+
"configuration.",
580+
_partitions_to_writers.size(), _partitions_to_writers.size() + 1,
575581
config::table_sink_partition_write_max_partition_nums_per_writer);
576582
}
577583
RETURN_IF_ERROR(create_and_open_writer(partition_name, i, nullptr, 0, writer));

‎be/src/exec/sink/writer/vhive_table_writer.cpp‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -176,8 +176,14 @@ Status VHiveTableWriter::write(RuntimeState* state, Block& block) {
176176
std::shared_ptr<VHivePartitionWriter> writer;
177177
if (_partitions_to_writers.size() + 1 >
178178
config::table_sink_partition_write_max_partition_nums_per_writer) {
179-
return Status::InternalError(
180-
"Too many open partitions {}",
179+
return Status::InvalidArgument(
180+
"Too many open partitions for Hive writer: open partitions={}, "
181+
"attempted partitions={}, limit={}. This is a per-writer limit "
182+
"controlled by BE configuration "
183+
"'table_sink_partition_write_max_partition_nums_per_writer'. "
184+
"Split the write into multiple statements or increase this "
185+
"configuration.",
186+
_partitions_to_writers.size(), _partitions_to_writers.size() + 1,
181187
config::table_sink_partition_write_max_partition_nums_per_writer);
182188
}
183189
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‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
#include <gtest/gtest.h>
19+
20+
#include "common/config.h"
21+
#include "core/block/block.h"
22+
#include "core/column/column_vector.h"
23+
#include "core/data_type/data_type_number.h"
24+
#include "exec/sink/writer/iceberg/viceberg_table_writer.h"
25+
#include "exec/sink/writer/vhive_table_writer.h"
26+
#include "exprs/vexpr_context.h"
27+
#include "exprs/vslot_ref.h"
28+
#include "runtime/runtime_state.h"
29+
30+
namespace doris {
31+
32+
class ExternalTableWriterPartitionLimitTest : public testing::TestWithParam<int32_t> {
33+
protected:
34+
void SetUp() override {
35+
_original_limit = config::table_sink_partition_write_max_partition_nums_per_writer;
36+
config::table_sink_partition_write_max_partition_nums_per_writer = GetParam();
37+
}
38+
39+
void TearDown() override {
40+
config::table_sink_partition_write_max_partition_nums_per_writer = _original_limit;
41+
}
42+
43+
Block new_partition_block() {
44+
auto column = ColumnInt32::create();
45+
column->get_data().push_back(GetParam());
46+
return Block(
47+
{ColumnWithTypeAndName(std::move(column), std::make_shared<DataTypeInt32>(), "p")});
48+
}
49+
50+
void check_limit_status(const Status& status, const std::string& writer_type) {
51+
EXPECT_EQ(TStatusCode::INVALID_ARGUMENT, status.to_thrift().status_code);
52+
const auto message = status.to_string();
53+
EXPECT_NE(std::string::npos, message.find(writer_type + " writer"));
54+
EXPECT_NE(std::string::npos, message.find("open partitions=" + std::to_string(GetParam())));
55+
EXPECT_NE(std::string::npos,
56+
message.find("attempted partitions=" + std::to_string(GetParam() + 1)));
57+
EXPECT_NE(std::string::npos, message.find("limit=" + std::to_string(GetParam())));
58+
EXPECT_NE(std::string::npos, message.find("per-writer limit"));
59+
EXPECT_NE(std::string::npos,
60+
message.find("table_sink_partition_write_max_partition_nums_per_writer"));
61+
EXPECT_NE(std::string::npos, message.find("Split the write into multiple statements"));
62+
}
63+
64+
private:
65+
int32_t _original_limit = 0;
66+
};
67+
68+
TEST_P(ExternalTableWriterPartitionLimitTest, HiveRejectsNextPartitionBeforeOpeningAFile) {
69+
THiveColumn partition_column;
70+
partition_column.__set_name("p");
71+
partition_column.__set_column_type(THiveColumnType::PARTITION_KEY);
72+
THiveTableSink hive_sink;
73+
hive_sink.__set_columns({partition_column});
74+
TDataSink sink;
75+
sink.__set_hive_table_sink(hive_sink);
76+
auto type = std::make_shared<DataTypeInt32>();
77+
VExprContextSPtrs exprs = {
78+
VExprContext::create_shared(VSlotRef::create_shared(0, 0, -1, type, "p"))};
79+
VHiveTableWriter writer(sink, exprs, nullptr, nullptr);
80+
writer._partition_columns_input_index = {0};
81+
// Existing writers are never dereferenced when rejecting a new partition. No filesystem is needed.
82+
for (int32_t i = 0; i < GetParam(); ++i) {
83+
writer._partitions_to_writers.emplace("p=" + std::to_string(i), nullptr);
84+
}
85+
RuntimeState state;
86+
auto block = new_partition_block();
87+
check_limit_status(writer.write(&state, block), "Hive");
88+
EXPECT_EQ(static_cast<size_t>(GetParam()), writer._partitions_to_writers.size());
89+
}
90+
91+
TEST_P(ExternalTableWriterPartitionLimitTest, IcebergRejectsNextPartitionBeforeOpeningAFile) {
92+
TDataSink sink;
93+
sink.__set_iceberg_table_sink(TIcebergTableSink());
94+
VExprContextSPtrs exprs;
95+
iceberg::PartitionField field(1, 1000, "p", "identity");
96+
VIcebergTableWriter writer(sink, exprs, nullptr, nullptr);
97+
writer._iceberg_partition_columns.emplace_back(
98+
field, TYPE_INT, 0, std::vector<size_t> {},
99+
std::make_unique<IdentityPartitionColumnTransform>(std::make_shared<DataTypeInt32>()));
100+
for (int32_t i = 0; i < GetParam(); ++i) {
101+
writer._partitions_to_writers.emplace("existing-" + std::to_string(i), nullptr);
102+
}
103+
auto block = new_partition_block();
104+
check_limit_status(writer.write_prepared_block(block), "Iceberg");
105+
EXPECT_EQ(static_cast<size_t>(GetParam()), writer._partitions_to_writers.size());
106+
}
107+
108+
INSTANTIATE_TEST_SUITE_P(DefaultAndConfiguredLimits, ExternalTableWriterPartitionLimitTest,
109+
testing::Values(128, 2));
110+
111+
} // namespace doris

‎regression-test/suites/external_table_p0/hive/write/test_hive_write_partition_limit.groovy‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
suite("test_hive_write_partition_limit",
19+
"p0,external,hive,external_docker,external_docker_hive,nonConcurrent") {
20+
String enabled = context.config.otherConfigs.get("enableHiveTest")
21+
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
22+
logger.info("disable hive test")
23+
return
24+
}
25+
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
26+
sql """drop catalog if exists test_hive_write_partition_limit"""
27+
setHivePrefix("hive2")
28+
String hmsPort = context.config.otherConfigs.get("hive2HmsPort")
29+
sql """create catalog test_hive_write_partition_limit properties (
30+
"type" = "hms",
31+
"hive.metastore.uris" = "thrift://${externalEnvIp}:${hmsPort}",
32+
"hadoop.username" = "hive"
33+
)"""
34+
hive_docker """drop table if exists hive_write_partition_limit"""
35+
hive_docker """create table hive_write_partition_limit (id int)
36+
partitioned by (p int) stored as parquet"""
37+
sql """use test_hive_write_partition_limit.default"""
38+
39+
// The limit belongs to each writer, so route all VALUES rows through one writer.
40+
sql """set parallel_pipeline_task_num = 1"""
41+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer": "2"]) {
42+
// Exactly two partitions are allowed; the next partition must fail with actionable context.
43+
sql """insert into hive_write_partition_limit values (0, 0), (1, 1)"""
44+
test {
45+
sql """insert into hive_write_partition_limit values (2, 2), (3, 3), (4, 4)"""
46+
exception "[INVALID_ARGUMENT]Too many open partitions for Hive writer: " +
47+
"open partitions=2, attempted partitions=3, limit=2."
48+
}
49+
}
50+
// The advertised configuration is effective: the same three-partition write can now succeed.
51+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer": "3"]) {
52+
sql """insert into hive_write_partition_limit values (2, 2), (3, 3), (4, 4)"""
53+
}
54+
}

‎regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_partition_limit.groovy‎

Lines changed: 60 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
suite("test_iceberg_write_partition_limit",
19+
"p0,external,iceberg,external_docker,external_docker_iceberg,nonConcurrent") {
20+
String enabled = context.config.otherConfigs.get("enableIcebergTest")
21+
if (enabled == null || !enabled.equalsIgnoreCase("true")) {
22+
logger.info("disable iceberg test")
23+
return
24+
}
25+
String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
26+
sql """drop catalog if exists test_iceberg_write_partition_limit"""
27+
String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
28+
String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
29+
sql """create catalog test_iceberg_write_partition_limit properties (
30+
"type" = "iceberg",
31+
"iceberg.catalog.type" = "rest",
32+
"uri" = "http://${externalEnvIp}:${restPort}",
33+
"s3.access_key" = "admin",
34+
"s3.secret_key" = "password",
35+
"s3.endpoint" = "http://${externalEnvIp}:${minioPort}",
36+
"s3.region" = "us-east-1"
37+
)"""
38+
sql """create database if not exists test_iceberg_write_partition_limit.partition_limit_db"""
39+
sql """use test_iceberg_write_partition_limit.partition_limit_db"""
40+
sql """drop table if exists iceberg_write_partition_limit"""
41+
sql """create table iceberg_write_partition_limit (id int, p int)
42+
partition by list (p) ()
43+
properties ("format-version" = "2", "write.format.default" = "parquet")"""
44+
45+
// The limit belongs to each writer, so route all VALUES rows through one writer.
46+
sql """set parallel_pipeline_task_num = 1"""
47+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer": "2"]) {
48+
// Exactly two partitions are allowed; the next partition must fail with actionable context.
49+
sql """insert into iceberg_write_partition_limit values (0, 0), (1, 1)"""
50+
test {
51+
sql """insert into iceberg_write_partition_limit values (2, 2), (3, 3), (4, 4)"""
52+
exception "[INVALID_ARGUMENT]Too many open partitions for Iceberg writer: " +
53+
"open partitions=2, attempted partitions=3, limit=2."
54+
}
55+
}
56+
// The advertised configuration is effective: the same three-partition write can now succeed.
57+
setBeConfigTemporary(["table_sink_partition_write_max_partition_nums_per_writer": "3"]) {
58+
sql """insert into iceberg_write_partition_limit values (2, 2), (3, 3), (4, 4)"""
59+
}
60+
}

0 commit comments

Comments
 (0)