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 6261edede1b [fix](be) Bound the Variant metadata that Iceberg and
Paimon writes copy into each row (#69301)
6261edede1b is described below
commit 6261edede1b83edde6e3e87b4ccb4b90976f7ca8
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Oct 23 08:43:08 2026 +0100
[fix](be) Bound the Variant metadata that Iceberg and Paimon writes copy
into each row (#69301)
### What problem does this PR solve?
Issue Number: None
Related PR: #68883, #69194
Problem Summary:
**In short.** An INSERT of JSON with many distinct keys into a VARIANT
column of an Iceberg or Paimon table fails with an Arrow capacity error,
or runs out of memory first, because the writers copy the metadata
dictionary of a whole batch into every row. This PR writes a row whose
shared dictionary is much larger than its value with a dictionary of
only its own keys, as the Flight path already does, and keeps small
shared dictionaries as they are.
**Background.** A Variant value is a metadata dictionary of keys plus a
value that refers to keys by their index. `parse_json()` encodes a block
of rows with one `JsonStringToVariantEncoder`, so all rows of the block
share one dictionary holding the keys of every row. Iceberg
(`write_column_to_iceberg_arrow`, which passes `compact_metadata =
false`) and Paimon (`write_binary_variant_arrow`) copy that dictionary
into the metadata of every row they write, and the Paimon writer
validates it once per row. The Flight path (#68883) instead re-encodes a
row that does not use its whole dictionary.
**The problem, and what it cost.** The bytes a batch writes into the
metadata column, and on Paimon the validation work, grow as rows ×
dictionary bytes. 4064 rows that each have a distinct 248-byte key share
a 1 MiB dictionary, so the metadata column needs about 4 GiB, past the 2
GiB an Arrow binary array holds. On a local cluster, `INSERT ... SELECT
number, PARSE_TO_VARIANT(<one such key per row>) FROM numbers("number" =
"8192")`:
- into a Paimon table failed with `Variant V2 Arrow append failed:
Capacity error: array cannot contain more than 2147483646 bytes, have
2148527183`;
- into an Iceberg v3 table hit the same Arrow error, which
`checkArrowStatus` turns into a `FatalError`, so a debug or ASAN BE
aborted (a release BE fails the INSERT).
Nothing was written. Below that limit the copy still costs memory, and
the Paimon validation is quadratic in the number of keys. branch-4.1 and
branch-4.2 have the same code (found by reading it).
**How this PR fixes it.** `with_row_metadata()` in
`data_type_variant_v2_serde.cpp` decides the dictionary of each row,
under one of two policies:
- `OWN_KEYS` (Flight, unchanged): a row that does not use its whole
dictionary is re-encoded with a dictionary of only its keys.
- `BOUNDED` (Iceberg and Paimon): a row keeps the dictionary it shares
with its batch unless that is more than 1 KiB larger than the row's
value; then it is re-encoded as under `OWN_KEYS`.
So a table write copies at most the value bytes plus 1 KiB of metadata a
row (or the row's own keys), and re-encoding a row costs no more than
copying the dictionary would have. A small dictionary stays shared, so
the rows keep identical metadata that Parquet dictionary-encodes, and
nothing changes for them. The Paimon writer now checks the size limits
and validates the dictionary of the row as written, so a re-encoded row
is validated against its own keys. Re-encoding changes the key indexes
in the value, not what the value means; readers decode each row with its
own metadata.
**Results.**
| Batch | Before | After |
|---|---|---|
| 256 rows, a distinct 200-byte key each (BE UT) | each row carries the
52 KiB dictionary (13 MiB) | each row carries its own key |
| A row whose 64 KiB value outweighs that dictionary | shared dictionary
| shared dictionary (unchanged) |
| Small shared dictionary | shared dictionary | shared dictionary, bytes
unchanged |
| INSERT of 8192 rows with distinct 248-byte keys into Iceberg v3 /
Paimon (regression) | Iceberg: Arrow capacity error (an ASAN BE aborts);
Paimon: Arrow capacity error | both succeed; Doris and Spark read every
row back as written |
### Release note
Fix INSERT into Iceberg and Paimon VARIANT columns failing with an Arrow
capacity error, or running out of memory, when a batch of JSON has many
distinct keys.
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- New `test_iceberg_write_variant_many_keys` and
`test_paimon_write_variant_many_keys`: the INSERT above, then Doris
checks every row against the JSON it wrote and Spark (`to_json`) does
the same. They pass on a local cluster with the iceberg docker
environment; on a BE built with master's serde the INSERTs fail as
described. The existing `test_iceberg_variant_read`,
`test_iceberg_variant_defaults` and the six `test_paimon_write_variant*`
suites pass.
- [x] Unit Test
- `DataTypeVariantV2SerdeOutputTest`:
`TableWritersGiveRowsTheirOwnKeysInsteadOfALargeDictionary` (Iceberg and
Paimon layouts; every row decodes to its key and value) and
`TableWritersKeepASmallSharedDictionary`. These and the related Variant
serde, Paimon convertor, Flight Variant, Arrow and Parquet writer tests
(32) pass. Each of these mutations turns the new tests red: Iceberg
compacting every row, keeping every shared dictionary, Paimon not
compacting, an absolute 1 KiB threshold instead of one relative to the
value, and treating table writers as Flight.
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
Co-authored-by: Claude Opus 5.5 (1M context) <[email protected]>
---
.../data_type_serde/data_type_variant_v2_serde.cpp | 78 +++++++++++----
.../data_type_variant_v2_serde_output_test.cpp | 109 +++++++++++++++++++++
.../write/test_iceberg_write_variant_many_keys.out | 4 +
.../write/test_paimon_write_variant_many_keys.out | 4 +
.../test_iceberg_write_variant_many_keys.groovy | 90 +++++++++++++++++
.../test_paimon_write_variant_many_keys.groovy | 89 +++++++++++++++++
6 files changed, 354 insertions(+), 20 deletions(-)
diff --git a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
index ef2d3a1536c..88361955030 100644
--- a/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_variant_v2_serde.cpp
@@ -102,6 +102,49 @@ Status append_flight_variant_value(VariantRef value,
VariantBatchBuilder::Row& o
return Status::OK();
}
+// The dictionary that a writer gives each Variant row; see
with_row_metadata().
+enum class VariantRowMetadata {
+ // Flight sends every row with a dictionary of only the keys it uses.
+ OWN_KEYS,
+ // Table writers keep the dictionary a row shares with its batch unless it
is more than
+ // SHARED_METADATA_SLACK bytes larger than the row's value.
+ BOUNDED,
+};
+
+// Rows encoded together, such as the rows of one parse_json() call, share one
dictionary of all
+// their keys, and a table writer copies the dictionary into every row. 4064
rows with a distinct
+// 250-byte key each share a 1 MiB dictionary, so copying it costs 4 GiB, past
the 2 GiB that an
+// Arrow binary array holds. A row whose dictionary exceeds its value by more
than this slack gets
+// its own keys instead, which bounds the copied metadata by the value bytes
plus 1 KiB a row. A
+// smaller dictionary is kept: the rows of a batch then carry the same
metadata, which Parquet
+// dictionary-encodes.
+constexpr size_t SHARED_METADATA_SLACK = 1024;
+
+// Re-encodes `value` with a dictionary of only the keys it uses, kept alive
by `owner`, when
+// `policy` asks for that and the dictionary holds keys the row does not use.
For example, under
+// BOUNDED, {"a": 1} keeps a shared dictionary [a, b] but gets [a] when the
dictionary also holds
+// 4000 other keys.
+Status with_row_metadata(VariantRef& value, VariantRowMetadata policy,
+ std::optional<VariantBatchBuilder>& owner) {
+ const auto keys = value.metadata.dict_size();
+ // An empty dictionary or an object using every key already has row-local
metadata.
+ if (keys == 0 ||
+ (value.basic_type() == VariantBasicType::OBJECT &&
value.num_elements() == keys)) {
+ return Status::OK();
+ }
+ if (policy == VariantRowMetadata::BOUNDED &&
+ value.metadata.size <= value.value.size + SHARED_METADATA_SLACK) {
+ return Status::OK();
+ }
+ VariantBatchBuilder encoder;
+ auto row = encoder.begin_row();
+ RETURN_IF_ERROR(append_flight_variant_value(value, row));
+ row.finish();
+ owner.emplace(encoder.finish_batch());
+ value = owner->value_at(0);
+ return Status::OK();
+}
+
using MetaIdsColumn = ColumnVector<TYPE_UINT32>;
std::span<const NullMap::value_type> forced_nulls(const NullMap* null_map) {
@@ -336,7 +379,11 @@ Status write_binary_variant_arrow(const IColumn& column,
const NullMap* null_map
column, start, end, outer_nulls,
[&](size_t) { require_variant_arrow_status(builder.AppendNull());
},
[&](size_t row, VariantRef value) {
+ std::optional<VariantBatchBuilder> own_keys;
try {
+ // Check and validate the row as it is written: a row
given its own keys is
+ // validated against them instead of the dictionary its
batch shares.
+ THROW_IF_ERROR(with_row_metadata(value,
VariantRowMetadata::BOUNDED, own_keys));
constexpr size_t BINARY_VARIANT_SIZE_LIMIT = 128 * 1024 *
1024;
if (value.value.size > BINARY_VARIANT_SIZE_LIMIT ||
value.metadata.size > BINARY_VARIANT_SIZE_LIMIT) {
@@ -784,7 +831,7 @@ Status write_arrow(const IColumn& column, const NullMap*
null_map, Builder& buil
Status write_parquet_variant_arrow(const IColumn& column, const NullMap*
null_map,
arrow::ArrayBuilder* array_builder, int64_t
start, int64_t end,
- bool compact_metadata) {
+ VariantRowMetadata row_metadata) {
if (start < 0 || end < start) {
return Status::InvalidArgument("Invalid Variant Arrow row range [{},
{})", start, end);
}
@@ -815,22 +862,10 @@ Status write_parquet_variant_arrow(const IColumn& column,
const NullMap* null_ma
if (!status.ok()) {
return;
}
- std::optional<VariantBatchBuilder> compacted;
- if (compact_metadata) {
- const auto keys = value.metadata.dict_size();
- // An empty dictionary or an object using every key
already has row-local metadata.
- if (keys != 0 && (value.basic_type() !=
VariantBasicType::OBJECT ||
- value.num_elements() != keys)) {
- VariantBatchBuilder encoder;
- auto row = encoder.begin_row();
- status = append_flight_variant_value(value, row);
- if (!status.ok()) {
- return;
- }
- row.finish();
- compacted.emplace(encoder.finish_batch());
- value = compacted->value_at(0);
- }
+ std::optional<VariantBatchBuilder> own_keys;
+ status = with_row_metadata(value, row_metadata, own_keys);
+ if (!status.ok()) {
+ return;
}
if (value.metadata.size > std::numeric_limits<int32_t>::max()
||
value.value.size > std::numeric_limits<int32_t>::max()) {
@@ -909,7 +944,8 @@ Status DataTypeVariantV2SerDe::write_column_to_arrow(const
IColumn& column, cons
const size_t first = checked_row(start);
const size_t last = checked_row(end);
if (is_native_arrow_type(*array_builder->type())) {
- return write_parquet_variant_arrow(column, null_map,
array_builder, start, end, true);
+ return write_parquet_variant_arrow(column, null_map,
array_builder, start, end,
+ VariantRowMetadata::OWN_KEYS);
}
if (array_builder->type()->id() == arrow::Type::STRING) {
return write_arrow(column, null_map,
assert_cast<arrow::StringBuilder&>(*array_builder),
@@ -952,9 +988,11 @@ Status
DataTypeVariantV2SerDe::write_column_to_iceberg_arrow(
const std::shared_ptr<const IDataType>&, const IColumn& column, const
NullMap* null_map,
const std::shared_ptr<arrow::Field>&, arrow::ArrayBuilder*
array_builder, int64_t start,
int64_t end, const cctz::time_zone&) const {
- // Table writers retain their validated dictionary; Flight compacts each
transmitted row.
+ // A row keeps a small dictionary shared with its batch and gets its own
keys instead of a
+ // large one; Flight compacts each transmitted row.
RETURN_IF_CATCH_EXCEPTION({
- return write_parquet_variant_arrow(column, null_map, array_builder,
start, end, false);
+ return write_parquet_variant_arrow(column, null_map, array_builder,
start, end,
+ VariantRowMetadata::BOUNDED);
});
return Status::OK();
}
diff --git
a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
index bf9be8ac2f2..8f02ede8b4b 100644
--- a/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
+++ b/be/test/core/data_type_serde/data_type_variant_v2_serde_output_test.cpp
@@ -28,6 +28,7 @@
#include <span>
#include <string>
#include <string_view>
+#include <utility>
#include <vector>
#include "core/arena.h"
@@ -216,6 +217,49 @@ std::unique_ptr<arrow::StructBuilder>
binary_variant_arrow_builder() {
std::make_shared<arrow::BinaryBuilder>(arrow::default_memory_pool())});
}
+// The struct<metadata, value> storage that Iceberg writes into Parquet.
+std::unique_ptr<arrow::StructBuilder> parquet_variant_arrow_builder() {
+ return std::make_unique<arrow::StructBuilder>(
+ arrow::struct_({arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false)}),
+ arrow::default_memory_pool(),
+ std::vector<std::shared_ptr<arrow::ArrayBuilder>> {
+
std::make_shared<arrow::BinaryBuilder>(arrow::default_memory_pool()),
+
std::make_shared<arrow::BinaryBuilder>(arrow::default_memory_pool())});
+}
+
+// The Variant rows that the Iceberg and the Paimon writer produce for
`column`, as they write them.
+std::vector<std::vector<std::pair<std::string, std::string>>>
table_variant_rows(
+ const IColumn& column) {
+ DataTypeVariantV2SerDe serde;
+ auto iceberg = parquet_variant_arrow_builder();
+ Status status =
+ serde.write_column_to_iceberg_arrow(nullptr, column, nullptr,
nullptr, iceberg.get(), 0,
+ column.size(),
cctz::utc_time_zone());
+ EXPECT_TRUE(status.ok()) << status;
+ auto paimon = binary_variant_arrow_builder();
+ status = serde.write_column_to_paimon_arrow(nullptr, column, nullptr,
nullptr, paimon.get(), 0,
+ column.size(),
cctz::utc_time_zone());
+ EXPECT_TRUE(status.ok()) << status;
+
+ std::vector<std::vector<std::pair<std::string, std::string>>> writers;
+ const std::array<std::pair<arrow::StructBuilder*, int>, 2> builders {
+ {{iceberg.get(), 0}, {paimon.get(), 1}}};
+ for (const auto& [builder, metadata_field] : builders) {
+ std::shared_ptr<arrow::Array> output;
+ EXPECT_TRUE(builder->Finish(&output).ok());
+ const auto& array = assert_cast<const arrow::StructArray&>(*output);
+ const auto& metadata = assert_cast<const
arrow::BinaryArray&>(*array.field(metadata_field));
+ const auto& values =
+ assert_cast<const arrow::BinaryArray&>(*array.field(1 -
metadata_field));
+ auto& rows = writers.emplace_back();
+ for (int64_t row = 0; row < array.length(); ++row) {
+ rows.emplace_back(std::string(metadata.GetView(row)),
std::string(values.GetView(row)));
+ }
+ }
+ return writers;
+}
+
void expect_binary_variant_bytes(const DataTypeVariantV2SerDe& serde, const
IColumn& column,
const ColumnVariantV2& encoded,
const NullMap* null_map = nullptr) {
@@ -509,6 +553,71 @@ TEST(DataTypeVariantV2SerdeOutputTest,
PaimonKeepsItsOwnBinaryLayout) {
EXPECT_EQ(binary.GetView(0), std::string_view(expected.data,
expected.size));
}
+TEST(DataTypeVariantV2SerdeOutputTest,
TableWritersGiveRowsTheirOwnKeysInsteadOfALargeDictionary) {
+ // parse_json() encodes a batch with one dictionary of all its keys: 256
rows with a distinct
+ // 200-byte key share a 52 KiB dictionary. Copied into every row it would
cost 13 MiB for 2 KiB of
+ // values, and 4064 such rows overflow an Arrow binary array, so the
Iceberg and the Paimon writer
+ // give each row a dictionary of its own key; the row still decodes to its
key and value. Row 0
+ // holds a 64 KiB string, which the shared dictionary does not exceed by
more than 1 KiB, so it
+ // keeps that dictionary. MUTATION: keeping every shared dictionary, or
giving row 0 its own key
+ // -> red.
+ constexpr size_t ROWS = 256;
+ JsonStringToVariantEncoder encoder(
+ {.max_json_key_length = 1024, .throw_on_invalid_json = true});
+ std::vector<std::string> keys;
+ for (size_t row = 0; row < ROWS; ++row) {
+ keys.push_back("k" + std::to_string(row) + std::string(200, 'x'));
+ const std::string value =
+ row == 0 ? "\"" + std::string(64 * 1024, 'v') + "\"" :
std::to_string(row);
+ const std::string document = "{\"" + keys.back() + "\":" + value + "}";
+ encoder.add_json({document.data(), document.size()});
+ }
+ VariantBatchBuilder batch = encoder.finish_batch();
+ auto column = ColumnVariantV2::create();
+ column->insert_encoded_batch(batch);
+ const VariantRef shared = column->get_value_ref(1);
+ ASSERT_EQ(shared.metadata.dict_size(), ROWS);
+
+ for (const auto& rows : table_variant_rows(*column)) {
+ ASSERT_EQ(rows.size(), ROWS);
+ EXPECT_EQ(rows[0].first, std::string_view(shared.metadata.data,
shared.metadata.size));
+ for (size_t row = 0; row < ROWS; ++row) {
+ const VariantRef value {
+ .metadata = {.data = rows[row].first.data(), .size =
rows[row].first.size()},
+ .value = {rows[row].second.data(),
rows[row].second.size()}};
+ if (row != 0) {
+ EXPECT_EQ(value.metadata.dict_size(), 1U) << row;
+ }
+ ASSERT_EQ(value.basic_type(), VariantBasicType::OBJECT) << row;
+ ASSERT_EQ(value.num_elements(), 1U) << row;
+ uint32_t field_id = 0;
+ const VariantRef child = value.object_view().value_at(0,
&field_id);
+ EXPECT_EQ(value.metadata.key_at(field_id).to_string(), keys[row]);
+ if (row == 0) {
+ EXPECT_EQ(child.get_string().size, size_t {64 * 1024});
+ } else {
+ EXPECT_EQ(child.get_int(), static_cast<int64_t>(row));
+ }
+ }
+ }
+}
+
+TEST(DataTypeVariantV2SerdeOutputTest, TableWritersKeepASmallSharedDictionary)
{
+ // A dictionary that exceeds a row's value by at most 1 KiB is written as
the batch shares it:
+ // Parquet dictionary-encodes the metadata the rows then have in common,
and the row is not
+ // re-encoded. MUTATION: giving every row its own keys, as Flight does ->
red.
+ auto column = encoded_json({R"({"a":1})", R"({"b":[true,"x"]})",
R"({"a":{"b":2}})"});
+ for (const auto& rows : table_variant_rows(*column)) {
+ ASSERT_EQ(rows.size(), column->size());
+ for (size_t row = 0; row < column->size(); ++row) {
+ const VariantRef expected = column->get_value_ref(row);
+ EXPECT_EQ(rows[row].first,
+ std::string_view(expected.metadata.data,
expected.metadata.size));
+ EXPECT_EQ(rows[row].second, std::string_view(expected.value.data,
expected.value.size));
+ }
+ }
+}
+
TEST(DataTypeVariantV2SerdeOutputTest,
BinaryStructRejectsUnsupportedPrimitive) {
DataTypeVariantV2SerDe serde;
VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows = 1});
diff --git
a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_variant_many_keys.out
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_variant_many_keys.out
new file mode 100644
index 00000000000..b596720ad0e
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_variant_many_keys.out
@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !many_keys_read_back --
+8192 8192
+
diff --git
a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_many_keys.out
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_many_keys.out
new file mode 100644
index 00000000000..b596720ad0e
--- /dev/null
+++
b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_many_keys.out
@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !many_keys_read_back --
+8192 8192
+
diff --git
a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_variant_many_keys.groovy
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_variant_many_keys.groovy
new file mode 100644
index 00000000000..9cca5f14e76
--- /dev/null
+++
b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_variant_many_keys.groovy
@@ -0,0 +1,90 @@
+// 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.
+
+// PARSE_TO_VARIANT encodes a block with one metadata dictionary of all the
keys of its rows. Here every row has
+// its own 248-byte key, so the dictionary of a 4064-row block is 1 MiB, and
copying it into the metadata of every
+// row would take 4 GiB, past the 2 GiB that an Arrow binary array holds. The
writer gives such a row a dictionary
+// of its own key instead, so the INSERT succeeds, and Doris and Spark read
every row back as it was written.
+suite("test_iceberg_write_variant_many_keys",
+ "p0,external,iceberg,external_docker,external_docker_iceberg") {
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test")
+ return
+ }
+
+ String restPort = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String restUri = context.config.otherConfigs.get("iceberg_rest_uri")
+ if (restUri == null) {
+ restUri = "http://${externalEnvIp}:${restPort}";
+ }
+ String catalogName = "test_iceberg_write_variant_many_keys"
+ String dbName = "iceberg_write_variant_many_keys_db"
+
+ spark_iceberg_multi """
+ CREATE NAMESPACE IF NOT EXISTS demo.${dbName};
+ DROP TABLE IF EXISTS demo.${dbName}.variant_many_keys;
+ CREATE TABLE demo.${dbName}.variant_many_keys (id BIGINT) USING iceberg
+ TBLPROPERTIES ('format-version'='3', 'write.format.default'='parquet');
+ ALTER TABLE demo.${dbName}.variant_many_keys ADD COLUMN payload
VARIANT;
+ """
+
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type'='iceberg',
+ 'iceberg.catalog.type'='rest',
+ 'uri'='${restUri}',
+ 's3.access_key'='admin',
+ 's3.secret_key'='password',
+ 's3.endpoint'='http://${externalEnvIp}:${minioPort}',
+ 's3.region'='us-east-1',
+ 'meta.cache.iceberg.table.ttl-second'='0',
+ 'meta.cache.iceberg.schema.ttl-second'='0'
+ )
+ """
+ sql """SWITCH ${catalogName}"""
+ sql """USE ${dbName}"""
+ sql """SET enable_file_scanner_v2=true"""
+
+ sql """
+ INSERT INTO variant_many_keys
+ SELECT number,
+ PARSE_TO_VARIANT(CONCAT('{"', REPEAT('k', 240),
LPAD(CAST(number AS STRING), 8, '0'), '":',
+ CAST(number AS STRING), '}'))
+ FROM numbers("number" = "8192")
+ """
+ order_qt_many_keys_read_back """
+ SELECT COUNT(*),
+ SUM(CASE WHEN CAST(payload AS STRING) = CONCAT('{"',
REPEAT('k', 240),
+ LPAD(CAST(id AS STRING), 8, '0'), '":', CAST(id AS
STRING), '}') THEN 1 ELSE 0 END)
+ FROM variant_many_keys
+ """
+
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_many_keys"""
+ List<List<Object>> sparkRows = spark_iceberg """
+ SELECT COUNT(*),
+ SUM(CASE WHEN to_json(payload) = CONCAT('{"', REPEAT('k', 240),
LPAD(CAST(id AS STRING), 8, '0'),
+ '":', CAST(id AS STRING), '}') THEN 1 ELSE 0 END)
+ FROM demo.${dbName}.variant_many_keys
+ """
+ assertEquals(1, sparkRows.size())
+ assertEquals("8192", sparkRows[0][0].toString())
+ assertEquals("8192", sparkRows[0][1].toString())
+}
diff --git
a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_many_keys.groovy
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_many_keys.groovy
new file mode 100644
index 00000000000..14134e796c4
--- /dev/null
+++
b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_many_keys.groovy
@@ -0,0 +1,89 @@
+// 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.
+
+// PARSE_TO_VARIANT encodes a block with one metadata dictionary of all the
keys of its rows. Here every row has
+// its own 248-byte key, so the dictionary of a 4064-row block is 1 MiB, and
copying it into the metadata of every
+// row would take 4 GiB, past the 2 GiB that an Arrow binary array holds,
after validating the whole dictionary once
+// per row. The writer gives such a row a dictionary of its own key instead,
so the INSERT succeeds, and Doris and
+// Spark read every row back as it was written.
+suite("test_paimon_write_variant_many_keys", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test.")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String catalogName = "test_pw_variant_many_keys_catalog"
+ String dbName = "test_pw_variant_many_keys_db"
+
+ spark_paimon_multi """
+ CREATE DATABASE IF NOT EXISTS paimon.${dbName};
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_many_keys;
+ CREATE TABLE paimon.${dbName}.t_variant_many_keys (
+ id BIGINT,
+ payload VARIANT
+ ) USING paimon
+ TBLPROPERTIES ('file.format' = 'parquet');
+ """
+
+ sql """DROP CATALOG IF EXISTS ${catalogName}"""
+ sql """
+ CREATE CATALOG ${catalogName} PROPERTIES (
+ 'type' = 'paimon',
+ 'paimon.catalog.type' = 'filesystem',
+ 'warehouse' = 's3://warehouse/wh',
+ 's3.endpoint' = 'http://${externalEnvIp}:${minioPort}',
+ 's3.access_key' = 'admin',
+ 's3.secret_key' = 'password',
+ 's3.path.style.access' = 'true'
+ )
+ """
+ sql """SWITCH ${catalogName}"""
+ sql """USE ${dbName}"""
+
+ try {
+ sql """SET force_jni_scanner = true"""
+ sql """
+ INSERT INTO t_variant_many_keys
+ SELECT number,
+ PARSE_TO_VARIANT(CONCAT('{"', REPEAT('k', 240),
LPAD(CAST(number AS STRING), 8, '0'), '":',
+ CAST(number AS STRING), '}'))
+ FROM numbers("number" = "8192")
+ """
+ sql """REFRESH TABLE t_variant_many_keys"""
+ order_qt_many_keys_read_back """
+ SELECT COUNT(*),
+ SUM(CASE WHEN CAST(payload AS STRING) = CONCAT('{"',
REPEAT('k', 240),
+ LPAD(CAST(id AS STRING), 8, '0'), '":', CAST(id AS
STRING), '}') THEN 1 ELSE 0 END)
+ FROM t_variant_many_keys
+ """
+ } finally {
+ sql """SET force_jni_scanner = false"""
+ }
+
+ List<List<Object>> sparkRows = spark_paimon """
+ SELECT COUNT(*),
+ SUM(CASE WHEN to_json(payload) = CONCAT('{"', REPEAT('k', 240),
LPAD(CAST(id AS STRING), 8, '0'),
+ '":', CAST(id AS STRING), '}') THEN 1 ELSE 0 END)
+ FROM paimon.${dbName}.t_variant_many_keys
+ """
+ assertEquals(1, sparkRows.size())
+ assertEquals("8192", sparkRows[0][0].toString())
+ assertEquals("8192", sparkRows[0][1].toString())
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
