You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2026/10/19 07:18:04 UTC
(doris) branch master updated: [feature](iceberg) Write Iceberg VARIANT columns and require NULL VARIANT defaults (#69194)
This is an automated email from the ASF dual-hosted git repository.
morningman 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 b9907e2041c [feature](iceberg) Write Iceberg VARIANT columns and require NULL VARIANT defaults (#69194)
b9907e2041c is described below
commit b9907e2041c467b352c91e27566e5bd3ebc1e8f1
Author: Mingyu Chen (Rayner) <mo...@gmail.com>
AuthorDate: Mon Oct 19 08:17:55 2026 +0100
[feature](iceberg) Write Iceberg VARIANT columns and require NULL VARIANT defaults (#69194)
### What problem does this PR solve?
Issue Number: None
Related PR: #66620, #67323
Problem Summary:
**In short.** Ports branch-4.1 #66620 (Iceberg Variant V2 writes) and
#67323 (NULL defaults for VARIANT) to master. On master Doris reads
Iceberg VARIANT columns but rejects every write to a table that has one,
and rejects VARIANT in Iceberg DDL. With this PR Doris writes VARIANT
values to format-version 3 Iceberg tables with Parquet data files:
top-level and nested values, through INSERT, INSERT OVERWRITE, UPDATE,
MERGE and `rewrite_data_files`. It also creates and alters top-level
VARIANT columns. For every connector sink, including Paimon, a write is
rejected when its source implicitly casts a VARIANT to a scalar. Before
this PR such a write silently stored object and array values as NULL.
Original author: @hubgeter
**Background.** Iceberg added the VARIANT type in format version 3 and
stores it in Parquet as a group of two binary leaves, `metadata` and
`value`. Master already reads such columns (#66413). For Paimon it
already encodes Variant V2 values to Arrow (#68320, #68883), and it
checks the nested shapes a connector Variant accepts in `CheckCast`. The
Iceberg side of the write path was never ported. Two other parts of
#66620 were also missing:
- the plan validator that rejects lossy common-type coercion into a
VARIANT target;
- the binding of a merge sink's output by position.
**The problem, and what it cost.**
| Statement on master | Result |
|---|---|
| `INSERT` / `UPDATE` / `MERGE` (with data) / `rewrite_data_files` on an
Iceberg table with a VARIANT column, at any depth | `Iceberg VARIANT
columns are read-only and cannot be written` |
| `CREATE TABLE ... (payload VARIANT) ENGINE=ICEBERG`, `ADD COLUMN ...
VARIANT`, `MODIFY COLUMN ... VARIANT` | `Unsupported type for Iceberg:
VARIANT` |
| `INSERT INTO paimon_t SELECT IF(TRUE, payload, 1) ...`, `UNION ALL` of
a Variant and a number, the same in `CASE`, a recursive CTE or `LATERAL
VIEW` | The common type is DECIMAL, so an object or array `payload` is
written as NULL with no error |
| `UPDATE iceberg_t SET operation = ...` on a table whose data column is
named `operation` | Merge sink outputs were matched by name, so this
column skipped the cast to its target type |
Even if the FE had let these writes through, the BE could not complete
them:
- The BE could not parse `variant` in the Iceberg schema JSON.
- It had no Arrow schema mapping for VARIANT.
- It would have published the Parquet statistics of the two Variant
leaves under the field id `-1`.
**How this PR fixes it.**
- **BE.**
- The Iceberg type system parses `variant`.
- The Iceberg Arrow schema maps VARIANT to the `arrow.parquet.variant`
extension over `struct<metadata: binary, value: binary>`. Master's
Variant encoder already fills it (#68883).
- The Iceberg Parquet writer counts a Variant's values and nulls on the
`metadata` leaf, under the Variant's own field id. It records no column
size and no bounds for the Variant.
- The format v2 Iceberg reader rejects a non-NULL VARIANT initial
default.
- **FE, Iceberg connector.**
- A top-level VARIANT maps to Iceberg VARIANT. A VARIANT nested in
STRUCT, ARRAY or MAP DDL is rejected.
- CREATE TABLE and CTAS need an effective format version of 3 and
Parquet data files. The format version comes from the table properties
and the catalog's `table-default` / `table-override`, and so does the
file format, with the `write-format` alias winning as it does for an
existing table.
- ADD COLUMN needs the same table settings. MODIFY of a VARIANT column
changes only its comment, nullability and position, so it also works on
v3 ORC tables. A change to or from VARIANT is rejected.
- Writes of data files whose target holds a VARIANT now need format
version 3, Parquet, and no `write.parquet.shred-variants=true`. This
replaces the read-only check. DELETE and delete-only MERGE are
unchanged.
- Column metrics are also collected for VARIANT fields.
- VARIANT write defaults, initial defaults and DDL defaults must be NULL
(#67323).
- **FE, engine.** `VariantWritePlanValidator` traces the lineage of each
VARIANT target column through projections, set operations, recursive
CTEs, generators, CTE consumers and scalar subqueries. It rejects an
implicit cast that turns a Variant into a non-Variant value. It is wired
once for every connector:
- in `BindSink`, for INSERT, INSERT OVERWRITE, CTAS, rewrite, and
changelog row-level DML (Paimon UPDATE / MERGE);
- in `BindExpression`, for the row-level merge sink (Iceberg UPDATE /
MERGE).
The merge sink now casts its data columns by position.
- **Differences from branch-4.x.**
- The validator's message has no `Iceberg` / `Paimon` prefix, because
one generic check serves both.
- It does not inspect the condition of an IF or of a CASE branch, or
columns read only by such a condition. Branch-4.x rejects
`IF(payload['kind'] = 'event', payload, NULL)`, because comparing a
Variant path with a string implicitly casts the path, even though the
written value is unchanged.
- It does not trace the partition and order keys of a window, or the
ORDER BY of an aggregate function such as GROUP_CONCAT, which only
arrange the rows whose values are used. Branch-4.x rejects `ROW_NUMBER()
OVER (ORDER BY IF(TRUE, payload, 1))` into a VARIANT target.
- It does not trace the later inputs of EXCEPT and INTERSECT, which only
decide which rows of the first input stay. Branch-4.x traces every input
of a set operation.
- It follows the work-table columns that the recursive term of a
recursive CTE reads, and the WITH clauses nested in a producer taken
from the analyzer context, which a leading WITH binds against. It traces
each output of a CTE producer shared by several consumers once,
whichever sets of outputs the consumers request. Branch-4.x lets a loss
through in the first two cases and walks a shared producer once for
every path through the plan.
- Paimon UPDATE and MERGE are validated too. Branch-4.x validates only
Paimon INSERT.
- A Parquet leaf without a field id now fails a check (`DORIS_CHECK_GE`)
instead of being skipped. Every field of the schema the file is written
with has one.
- **Not ported.** Legacy Variant handling and the `enable_variant_v2`
switch (master is V2 only), the VALUES and MERGE action special cases
(master already casts each value and each branch to the target type),
and the extension and Arrow pool helpers (master has equivalents).
**Results.** These are from a local single-BE cluster with the Iceberg
REST and Paimon docker environments. The tables were created by Spark
4.0 with Iceberg 1.10.1, or by Doris DDL.
| Case | master | with this PR |
|---|---|---|
| `CREATE TABLE ... (payload VARIANT) ENGINE=ICEBERG` with
`format-version` 3 and Parquet | `Unsupported type for Iceberg: VARIANT`
| created. With format version 2: `Iceberg VARIANT writes require table
format-version 3, but found 2`. With ORC: `... require Parquet data
files, but found ORC`. `STRUCT<..., payload: VARIANT>`: rejected |
| INSERT of 13 values into that table (objects, arrays, primitives, JSON
null, SQL NULL, `"null"`, `""`, `{}`, `[]`) | read-only | written. Spark
reads each value back through `variant_get` and `to_json`, with SQL NULL
and JSON null kept apart. The data files report 13 values, 1 null, and
no bounds for the column |
| One writer inserts 100,000 rows into `STRUCT<label, payload VARIANT>`
| read-only | One data file with several row groups. Spark's `.files`
shows 100,000 values and 10,000 nulls under the nested field id, with no
column size and no bounds |
| Variant nested in STRUCT, ARRAY and MAP, and four levels deep |
read-only | Written. Spark checks the container shapes, Doris reads the
nested values back |
| `UPDATE` / `MERGE` (with `WHEN NOT MATCHED INSERT`) on a Variant
column named `operation` | read-only | Written, and Spark reads the new
values |
| MERGE that keeps a Variant in one branch and inserts the number `1` in
another | read-only | The object stays intact and the number is written
as the Variant `1` |
| A table with `write.parquet.shred-variants=true`, an ORC v3 table |
read-only | Rejected with no new snapshot. `MODIFY COLUMN v VARIANT
COMMENT ...` on the ORC table still works |
| INSERT into Paimon of `UNION ALL` of a Variant and `1`, `IF(TRUE,
payload, 1)`, the same in `CASE`, a recursive CTE, `LATERAL VIEW` |
Objects and arrays are written as NULL (master's own suite asserted
this) | All 8 rejected: `VARIANT write cannot safely convert input
column 'payload': source expression implicitly casts VARIANT to
DECIMALV3(38, 9) before sink analysis`. The same 8 statements against
Iceberg are rejected too |
| Paimon `UPDATE ... SET payload = IF(TRUE, payload, 1)`, a MERGE whose
source is such a UNION, and the same UPDATE / MERGE on Iceberg | — |
Rejected |
| `IF(payload['kind'] = 'object', ..., payload)` in a Paimon UPDATE, and
`CASE WHEN payload['kind'] = ... THEN payload END` in an INSERT | — |
Written (branch-4.x rejects these) |
| Iceberg table that gained a VARIANT column through Spark: the column
omitted, `DEFAULT`, `DEFAULT(payload)` in UPDATE, MERGE and `INSERT ...
SELECT` | read-only | Every such row reads NULL. `ADD COLUMN ... VARIANT
DEFAULT NULL` works. A non-NULL default and `NOT NULL DEFAULT NULL` are
rejected, and the schema is left unchanged |
- **BE UT.** 10 cases are ported from branch-4.x, and
`IcebergBuildsItsOwnSchemaAndMetadata` goes back to its Variant form:
- `ArrowSchemaUtilTest.test_variant_field` /
`test_nested_variant_fields`
- `SchemaParserTest.parse_variant`
- `VariantTypeTest.to_string`
- `PrimitiveTypesTest.from_primitive_string`
- `VParquetWriterTest.WritesIcebergVariantAndCollectsLogicalMetrics` /
`WritesNestedIcebergVariant`
- `IcebergV2ReaderTest.AnnotateVariantInitialDefaultMustBeNull` /
`StructInitialDefaultKeepsVariantChildNull`
All 474 cases of the related suites pass under ASAN. Two mutations each
turn only their own two cases red: dropping the Variant metrics rule,
and accepting a non-NULL initial default. clang-tidy reports nothing on
the changed files.
- **FE UT.**
- `VariantWritePlanValidatorTest`: the 13 branch-4.x cases, plus eight
new ones. They show that IF and CASE conditions are not checked, that
columns read only by a condition are not traced, that the later inputs
of EXCEPT and INTERSECT are not traced, and that the keys of a window
and the ORDER BY of GROUP_CONCAT are not traced while a lossy cast in a
function argument is still rejected. They also show that a loss is
rejected when a recursive CTE carries it through another work-table
column, and when it sits in a WITH nested in a producer from the
analyzer context. Two validate chains of CTEs within a timeout: 30
levels that each read the previous one twice, and 24 levels whose
requested sets of outputs double at each level.
- `BindExpressionTest`: the positional merge-output cast, and the
validator on a Variant column named `operation`.
- `ExternalRowLevelMergePlanBuilderTest`: each MERGE branch is cast to
the Variant target before the branch is selected.
- Connector tests:
- `IcebergWritePlanProviderTest`: the write capability matrix; a full
`planWrite` of a v3 Parquet, ORC and shredding table; VARIANT write
defaults.
- `IcebergSchemaBuilderTest`: top-level and nested VARIANT DDL, the
effective file format precedence, and DDL defaults.
- `CatalogBackedIcebergCatalogOpsColumnEvolutionTest`: ADD COLUMN on v2,
ORC and Parquet tables, metadata-only MODIFY, and rejected type changes.
- `IcebergConnectorMetadataDdlTest`: CREATE TABLE with VARIANT.
- `IcebergSchemaUtilsTest`: the initial default.
Mutations turn their own cases red:
- in the validator, inspecting conditions, tracing condition columns,
tracing the keys of a window or the ORDER BY of an aggregate function,
and caching CTE traces per set of requested outputs or not at all;
restoring the validator from before the review fixes turns exactly the
other six new cases red;
- in the connector, collecting metrics only for primitive fields,
dropping the CREATE check, dropping the ADD COLUMN check, and dropping
the VARIANT MODIFY branch.
All 1580 fe-connector-iceberg cases pass, and so do 137 cases in 12
fe-core classes, including the existing sink-binding, changelog and
merge builders.
- **Regression.**
- `test_iceberg_variant_read` carries the branch-4.x write block,
without the legacy-Variant cases and the `enable_variant_v2` switch,
which master does not have. Its `.out` gains rows 3–5 of
`variant_nested_projection` and `variant_nested_expressions`, the same
rows as branch-4.2.
- The new `test_iceberg_variant_defaults` produces the same `.out` as
branch-4.2.
- In `test_paimon_write_variant_errors`, the 8 common-type cases are
back to `test { exception }`, and Paimon UPDATE / MERGE cases are added.
- All 72 suites under `external_table_p0/paimon/write` and
`external_table_p0/iceberg/write`, and the Iceberg row-lineage DML
suites, pass on the same cluster. The one exception is
`test_paimon_write_thread_lifecycle`, which needs Linux `/proc`.
Co-authored-by: daidai <ch...@selectdb.com>
---
be/src/exec/sink/viceberg_merge_sink.cpp | 5 +-
be/src/format/table/iceberg/arrow_schema_util.cpp | 11 +-
be/src/format/table/iceberg/types.cpp | 2 +
be/src/format/table/iceberg/types.h | 10 +
.../format/transformer/viceberg_parquet_writer.cpp | 47 +-
be/src/format_v2/table/iceberg_reader.cpp | 8 +
.../format/arrow/arrow_block_convertor_test.cpp | 8 +-
.../table/iceberg/arrow_schema_util_test.cpp | 97 +++
.../format/table/iceberg/schema_parser_test.cpp | 19 +
be/test/format/table/iceberg/type_test.cpp | 7 +
.../format/transformer/vparquet_writer_test.cpp | 161 +++++
be/test/format_v2/table/iceberg_reader_test.cpp | 102 ++++
.../doris/connector/iceberg/IcebergCatalogOps.java | 20 +-
.../iceberg/IcebergConnectorMetadata.java | 9 +-
.../connector/iceberg/IcebergSchemaBuilder.java | 58 +-
.../connector/iceberg/IcebergSchemaUtils.java | 3 +
.../iceberg/IcebergWritePlanProvider.java | 37 +-
.../iceberg/IcebergWriteSchemaContext.java | 2 +
.../connector/iceberg/IcebergWriterHelper.java | 34 +-
...BackedIcebergCatalogOpsColumnEvolutionTest.java | 76 +++
.../iceberg/IcebergConnectorMetadataDdlTest.java | 47 ++
.../iceberg/IcebergSchemaBuilderTest.java | 68 +++
.../connector/iceberg/IcebergSchemaUtilsTest.java | 10 +
.../iceberg/IcebergWritePlanProviderTest.java | 147 +++--
.../datasource/VariantWritePlanValidator.java | 529 ++++++++++++++++
.../nereids/rules/analysis/BindExpression.java | 116 ++--
.../doris/nereids/rules/analysis/BindSink.java | 8 +
.../apache/doris/nereids/types/VariantType.java | 16 +
.../datasource/VariantWritePlanValidatorTest.java | 625 +++++++++++++++++++
.../nereids/rules/analysis/BindExpressionTest.java | 65 ++
.../ExternalRowLevelMergePlanBuilderTest.java | 22 +
.../iceberg/test_iceberg_variant_defaults.out | 23 +
.../iceberg/test_iceberg_variant_read.out | 6 +
.../write/test_paimon_write_variant_errors.out | 6 +
.../iceberg/test_iceberg_variant_defaults.groovy | 156 +++++
.../iceberg/test_iceberg_variant_read.groovy | 676 ++++++++++++++++++++-
.../write/test_paimon_write_variant_errors.groovy | 197 +++---
37 files changed, 3230 insertions(+), 203 deletions(-)
diff --git a/be/src/exec/sink/viceberg_merge_sink.cpp b/be/src/exec/sink/viceberg_merge_sink.cpp
index b025ce2cb83..4548f0449b6 100644
--- a/be/src/exec/sink/viceberg_merge_sink.cpp
+++ b/be/src/exec/sink/viceberg_merge_sink.cpp
@@ -168,8 +168,9 @@ Status VIcebergMergeSink::write(RuntimeState* state, Block& block) {
_delete_row_count += delete_rows;
_insert_row_count += insert_rows;
- // A delete-only plan deliberately omits the data writer so Variant target schemas never enter
- // the unsupported Iceberg data-write path. Reject a mismatched FE plan before dereferencing it.
+ // A delete-only plan omits the data writer, which it does not need and which a BE before Variant
+ // writes could not open for a Variant target schema. Reject a mismatched FE plan before
+ // dereferencing it.
if (has_insert && !_writes_data_files) {
return Status::InternalError(
"Iceberg delete-only merge sink received a data insert operation");
diff --git a/be/src/format/table/iceberg/arrow_schema_util.cpp b/be/src/format/table/iceberg/arrow_schema_util.cpp
index 18aec26888f..316549c4a66 100644
--- a/be/src/format/table/iceberg/arrow_schema_util.cpp
+++ b/be/src/format/table/iceberg/arrow_schema_util.cpp
@@ -17,6 +17,7 @@
#include "format/table/iceberg/arrow_schema_util.h"
+#include <arrow/extension/parquet_variant.h>
#include <arrow/extension/uuid.h>
#include <arrow/type.h>
#include <arrow/util/key_value_metadata.h>
@@ -29,7 +30,8 @@ const char* IcebergArrowSchemaConvertor::MAP_TYPE_VALUE = "mapType";
const char* IcebergArrowSchemaConvertor::UUID_TYPE_VALUE = "uuid";
Status IcebergArrowSchemaConvertor::get_arrow_schema(std::shared_ptr<arrow::Schema>* result) const {
- // Field IDs, UUID annotations and timestamp bindings come from the pinned table schema.
+ // Field IDs, UUID annotations, Variant storage and timestamp bindings come from the pinned
+ // table schema.
std::vector<std::shared_ptr<arrow::Field>> fields;
RETURN_IF_ERROR(convert_fields(fields));
auto schema = arrow::schema(std::move(fields));
@@ -108,6 +110,13 @@ Status IcebergArrowSchemaConvertor::convert_to_arrow_field(
arrow_type = arrow::binary();
break;
+ case iceberg::TypeID::VARIANT:
+ arrow_type = arrow::extension::variant(arrow::struct_({
+ arrow::field("metadata", arrow::binary(), false),
+ arrow::field("value", arrow::binary(), false),
+ }));
+ break;
+
case iceberg::TypeID::UUID:
metadata[ORIGINAL_TYPE] = UUID_TYPE_VALUE;
arrow_type = _use_uuid_extension ? arrow::extension::uuid() : arrow::fixed_size_binary(16);
diff --git a/be/src/format/table/iceberg/types.cpp b/be/src/format/table/iceberg/types.cpp
index 7046e3b4f6f..8a40a716b93 100644
--- a/be/src/format/table/iceberg/types.cpp
+++ b/be/src/format/table/iceberg/types.cpp
@@ -174,6 +174,8 @@ std::unique_ptr<PrimitiveType> Types::from_primitive_string(const std::string& t
return std::make_unique<UUIDType>();
} else if (lower_type_string == "binary") {
return std::make_unique<BinaryType>();
+ } else if (lower_type_string == "variant") {
+ return std::make_unique<VariantType>();
} else {
std::regex fixed(R"(fixed\[\s*(\d+)\s*\])");
std::regex decimal(R"(decimal\(\s*(\d+)\s*,\s*(\d+)\s*\))");
diff --git a/be/src/format/table/iceberg/types.h b/be/src/format/table/iceberg/types.h
index b4801dd0a5e..9b8fd4dc0f6 100644
--- a/be/src/format/table/iceberg/types.h
+++ b/be/src/format/table/iceberg/types.h
@@ -48,6 +48,7 @@ enum TypeID {
UUID,
FIXED,
BINARY,
+ VARIANT,
DECIMAL,
STRUCT,
LIST,
@@ -279,6 +280,15 @@ public:
std::string to_string() const override { return "binary"; }
};
+class VariantType : public PrimitiveType {
+public:
+ ~VariantType() override = default;
+
+ TypeID type_id() const override { return TypeID::VARIANT; }
+
+ std::string to_string() const override { return "variant"; }
+};
+
class FixedType : public PrimitiveType {
public:
~FixedType() override = default;
diff --git a/be/src/format/transformer/viceberg_parquet_writer.cpp b/be/src/format/transformer/viceberg_parquet_writer.cpp
index ceff37500a5..5b951ec77b1 100644
--- a/be/src/format/transformer/viceberg_parquet_writer.cpp
+++ b/be/src/format/transformer/viceberg_parquet_writer.cpp
@@ -21,6 +21,8 @@
#include <parquet/schema.h>
#include <algorithm>
+#include <optional>
+#include <unordered_set>
#include "format/table/iceberg/iceberg_arrow_block_convertor.h"
#include "format/table/parquet_utils.h"
@@ -51,6 +53,33 @@ static std::string encode_iceberg_bound(const ::parquet::Statistics& stats, std:
encoded.erase(0, start);
return encoded;
}
+
+namespace {
+
+// The Iceberg field whose metrics a Parquet leaf column feeds.
+struct IcebergMetricColumn {
+ int field_id;
+ bool is_variant_metadata;
+};
+
+// The leaves of a Variant carry no Iceberg field id. Its metadata leaf counts the values and nulls
+// of the Variant field, and its value leaf feeds no metric: its byte statistics are no logical
+// Variant statistics.
+std::optional<IcebergMetricColumn> iceberg_metric_column(const ::parquet::schema::Node& leaf) {
+ const auto* parent = leaf.parent();
+ const bool is_variant_leaf = parent != nullptr && parent->logical_type() != nullptr &&
+ parent->logical_type()->is_variant();
+ if (is_variant_leaf && leaf.name() != "metadata") {
+ return std::nullopt;
+ }
+ const int field_id = is_variant_leaf ? parent->field_id() : leaf.field_id();
+ // Every field of the Iceberg schema the file is written with carries its id.
+ DORIS_CHECK_GE(field_id, 0);
+ return IcebergMetricColumn {field_id, is_variant_leaf};
+}
+
+} // namespace
+
VIcebergParquetWriter::VIcebergParquetWriter(RuntimeState* state, io::FileWriter* file_writer,
const VExprContextSPtrs& output_vexpr_ctxs,
std::vector<std::string> column_names,
@@ -97,16 +126,28 @@ Status VIcebergParquetWriter::collect_file_statistics_after_close(TIcebergColumn
std::map<int, std::string> lower_bounds;
std::map<int, std::string> upper_bounds;
std::map<int, std::shared_ptr<::parquet::Statistics>> merged_column_stats;
+ // Variant fields get value and null counts, but no column size and no bounds.
+ std::unordered_set<int> variant_field_ids;
const int num_row_groups = file_metadata->num_row_groups();
const int num_columns = file_metadata->num_columns();
for (int col_idx = 0; col_idx < num_columns; ++col_idx) {
- auto field_id = file_metadata->schema()->Column(col_idx)->schema_node()->field_id();
+ const auto metric_column =
+ iceberg_metric_column(*file_metadata->schema()->Column(col_idx)->schema_node());
+ if (!metric_column.has_value()) {
+ continue;
+ }
+ const auto [field_id, is_variant_metadata] = *metric_column;
+ if (is_variant_metadata) {
+ variant_field_ids.insert(field_id);
+ }
for (int rg_idx = 0; rg_idx < num_row_groups; ++rg_idx) {
auto row_group = file_metadata->RowGroup(rg_idx);
auto column_chunk = row_group->ColumnChunk(col_idx);
- column_sizes[field_id] += column_chunk->total_compressed_size();
+ if (!is_variant_metadata) {
+ column_sizes[field_id] += column_chunk->total_compressed_size();
+ }
if (column_chunk->is_stats_set()) {
auto column_stat = column_chunk->statistics();
@@ -129,7 +170,7 @@ Status VIcebergParquetWriter::collect_file_statistics_after_close(TIcebergColumn
null_value_counts[field_id] = null_count;
value_counts[field_id] += null_count;
}
- if (column_stat->HasMinMax()) {
+ if (!variant_field_ids.contains(field_id) && column_stat->HasMinMax()) {
has_any_min_max = true;
lower_bounds[field_id] = encode_iceberg_bound(*column_stat, column_stat->EncodeMin());
upper_bounds[field_id] = encode_iceberg_bound(*column_stat, column_stat->EncodeMax());
diff --git a/be/src/format_v2/table/iceberg_reader.cpp b/be/src/format_v2/table/iceberg_reader.cpp
index 87924be5966..1068567076b 100644
--- a/be/src/format_v2/table/iceberg_reader.cpp
+++ b/be/src/format_v2/table/iceberg_reader.cpp
@@ -524,6 +524,10 @@ static Status build_v2_json_scalar_default(const format::ColumnDefinition& field
const rapidjson::Value& json_value,
std::deque<std::string>* binary_storage, Field* result) {
const auto primitive_type = value_type->get_primitive_type();
+ if (primitive_type == TYPE_VARIANT) {
+ return Status::InvalidArgument(
+ "Iceberg VARIANT initial-default for field '{}' must be NULL", field.name);
+ }
std::string serialized_value = iceberg_json_scalar_text(json_value);
const bool binary_like =
field.initial_default_value_is_base64 || primitive_type == TYPE_VARBINARY;
@@ -606,6 +610,10 @@ static Status build_v2_initial_default_field(const format::ColumnDefinition& fie
const auto value_type = remove_nullable(data_type);
const auto primitive_type = value_type->get_primitive_type();
+ if (primitive_type == TYPE_VARIANT) {
+ return Status::InvalidArgument(
+ "Iceberg VARIANT initial-default for field '{}' must be NULL", field.name);
+ }
if (is_complex_type(primitive_type)) {
rapidjson::Document document;
document.Parse(field.initial_default_value->data(), field.initial_default_value->size());
diff --git a/be/test/format/arrow/arrow_block_convertor_test.cpp b/be/test/format/arrow/arrow_block_convertor_test.cpp
index a2c038d774b..f3db16d59de 100644
--- a/be/test/format/arrow/arrow_block_convertor_test.cpp
+++ b/be/test/format/arrow/arrow_block_convertor_test.cpp
@@ -18,6 +18,7 @@
#include "format/arrow/arrow_block_convertor.h"
#include <arrow/api.h>
+#include <arrow/extension_type.h>
#include <arrow/io/api.h>
#include <arrow/ipc/api.h>
#include <gtest/gtest.h>
@@ -79,13 +80,16 @@ TEST_F(ArrowBlockConvertorTest, ParquetRejectsMismatchedColumnNames) {
TEST_F(ArrowBlockConvertorTest, IcebergBuildsItsOwnSchemaAndMetadata) {
const std::string json =
- R"({"type":"struct","fields":[{"id":7,"name":"payload","required":false,"type":"string"}]})";
+ R"({"type":"struct","fields":[{"id":7,"name":"payload","required":false,"type":"variant"}]})";
auto schema = iceberg::SchemaParser::from_json(json);
iceberg::IcebergArrowBlockConvertor converter(*schema, &json, "UTC", cctz::utc_time_zone());
ASSERT_TRUE(converter.init().ok());
const auto& arrow_schema = converter.arrow_schema();
ASSERT_NE(nullptr, arrow_schema);
- EXPECT_EQ(arrow::Type::STRING, arrow_schema->field(0)->type()->id());
+ ASSERT_EQ(arrow::Type::EXTENSION, arrow_schema->field(0)->type()->id());
+ const auto& variant = static_cast<const arrow::ExtensionType&>(*arrow_schema->field(0)->type());
+ EXPECT_EQ(arrow::Type::STRUCT, variant.storage_type()->id());
+ EXPECT_EQ("arrow.parquet.variant", variant.extension_name());
ASSERT_NE(nullptr, arrow_schema->metadata());
EXPECT_EQ(json, arrow_schema->metadata()->Get("iceberg.schema").ValueOrDie());
EXPECT_EQ("7", arrow_schema->field(0)->metadata()->Get("PARQUET:field_id").ValueOrDie());
diff --git a/be/test/format/table/iceberg/arrow_schema_util_test.cpp b/be/test/format/table/iceberg/arrow_schema_util_test.cpp
index da85ac4b764..9d84b9e766b 100644
--- a/be/test/format/table/iceberg/arrow_schema_util_test.cpp
+++ b/be/test/format/table/iceberg/arrow_schema_util_test.cpp
@@ -18,6 +18,7 @@
#include "format/table/iceberg/arrow_schema_util.h"
#include <arrow/api.h>
+#include <arrow/extension/parquet_variant.h>
#include <arrow/io/api.h>
#include <arrow/status.h>
#include <arrow/type.h>
@@ -28,6 +29,7 @@
#include <parquet/arrow/writer.h>
#include <parquet/schema.h>
+#include "core/assert_cast.h"
#include "core/block/block.h"
#include "core/column/column_array.h"
#include "core/column/column_nullable.h"
@@ -358,6 +360,101 @@ TEST(ArrowSchemaUtilTest, test_binary_field_types) {
EXPECT_EQ("uuid", fields[3]->metadata()->Get("originalType").ValueUnsafe());
}
+TEST(ArrowSchemaUtilTest, test_variant_field) {
+ std::vector<NestedField> nested_fields;
+ nested_fields.emplace_back(true, 21, "payload", std::make_unique<VariantType>(), std::nullopt);
+ Schema schema(1, std::move(nested_fields));
+
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ Status st = IcebergArrowSchemaConvertor(schema, "utc").convert_fields(fields);
+ ASSERT_TRUE(st.ok()) << st;
+ ASSERT_EQ(1, fields.size());
+ EXPECT_EQ("21", fields[0]->metadata()->Get(pfid).ValueUnsafe());
+ ASSERT_EQ(arrow::Type::EXTENSION, fields[0]->type()->id());
+
+ const auto& extension = static_cast<const arrow::ExtensionType&>(*fields[0]->type());
+ EXPECT_EQ("arrow.parquet.variant", extension.extension_name());
+ ASSERT_EQ(arrow::Type::STRUCT, extension.storage_type()->id());
+ ASSERT_EQ(2, extension.storage_type()->num_fields());
+ EXPECT_EQ("metadata", extension.storage_type()->field(0)->name());
+ EXPECT_EQ(arrow::Type::BINARY, extension.storage_type()->field(0)->type()->id());
+ EXPECT_FALSE(extension.storage_type()->field(0)->nullable());
+ EXPECT_FALSE(extension.storage_type()->field(0)->HasMetadata());
+ EXPECT_EQ("value", extension.storage_type()->field(1)->name());
+ EXPECT_EQ(arrow::Type::BINARY, extension.storage_type()->field(1)->type()->id());
+ EXPECT_FALSE(extension.storage_type()->field(1)->nullable());
+ EXPECT_FALSE(extension.storage_type()->field(1)->HasMetadata());
+}
+
+TEST(ArrowSchemaUtilTest, test_nested_variant_fields) {
+ const std::string schema_json = R"({
+ "type": "struct",
+ "fields": [
+ {
+ "id": 10,
+ "name": "info",
+ "required": false,
+ "type": {
+ "type": "struct",
+ "fields": [
+ {"id": 11, "name": "payload", "required": false, "type": "variant"}
+ ]
+ }
+ },
+ {
+ "id": 20,
+ "name": "events",
+ "required": false,
+ "type": {
+ "type": "list",
+ "element-id": 21,
+ "element": "variant",
+ "element-required": false
+ }
+ },
+ {
+ "id": 30,
+ "name": "attrs",
+ "required": false,
+ "type": {
+ "type": "map",
+ "key-id": 31,
+ "key": "string",
+ "value-id": 32,
+ "value": "variant",
+ "value-required": false
+ }
+ }
+ ]
+ })";
+ std::unique_ptr<Schema> schema = SchemaParser::from_json(schema_json);
+
+ std::vector<std::shared_ptr<arrow::Field>> fields;
+ Status status = IcebergArrowSchemaConvertor(*schema, "UTC").convert_fields(fields);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(3, fields.size());
+
+ const auto& info = assert_cast<const arrow::StructType&>(*fields[0]->type());
+ ASSERT_EQ(arrow::Type::EXTENSION, info.field(0)->type()->id());
+ EXPECT_EQ("11", info.field(0)->metadata()->Get(pfid).ValueUnsafe());
+
+ const auto& events = assert_cast<const arrow::ListType&>(*fields[1]->type());
+ ASSERT_EQ(arrow::Type::EXTENSION, events.value_type()->id());
+ EXPECT_EQ("21", events.value_field()->metadata()->Get(pfid).ValueUnsafe());
+
+ const auto& attrs = assert_cast<const arrow::MapType&>(*fields[2]->type());
+ ASSERT_EQ(arrow::Type::EXTENSION, attrs.item_type()->id());
+ EXPECT_EQ("32", attrs.item_field()->metadata()->Get(pfid).ValueUnsafe());
+
+ for (const auto& variant_type :
+ {info.field(0)->type(), events.value_type(), attrs.item_type()}) {
+ const auto& extension = static_cast<const arrow::ExtensionType&>(*variant_type);
+ EXPECT_EQ("arrow.parquet.variant", extension.extension_name());
+ EXPECT_EQ("struct<metadata: binary not null, value: binary not null>",
+ extension.storage_type()->ToString());
+ }
+}
+
TEST(ArrowSchemaUtilTest, test_parquet_filed_id) {
std::string test_dir = "ut_dir/test_parquet_filed_id";
Status st;
diff --git a/be/test/format/table/iceberg/schema_parser_test.cpp b/be/test/format/table/iceberg/schema_parser_test.cpp
index 427e0e3c05c..f48afd53470 100644
--- a/be/test/format/table/iceberg/schema_parser_test.cpp
+++ b/be/test/format/table/iceberg/schema_parser_test.cpp
@@ -49,6 +49,18 @@ const std::string valid_struct_json = R"({
]
})";
+const std::string variant_struct_json = R"({
+ "type": "struct",
+ "fields": [
+ {
+ "id": 30,
+ "name": "payload",
+ "type": "variant",
+ "required": false
+ }
+ ]
+})";
+
const std::string invalid_json = R"({
"type": "struct",
"fields": [
@@ -202,6 +214,13 @@ TEST(SchemaParserTest, parse_valid_struct) {
EXPECT_EQ(age_field->field_type()->as_primitive_type()->to_string(), "int");
}
+TEST(SchemaParserTest, parse_variant) {
+ std::unique_ptr<Schema> schema = SchemaParser::from_json(variant_struct_json);
+ ASSERT_NE(schema, nullptr);
+ ASSERT_NE(schema->find_field(30), nullptr);
+ EXPECT_EQ(schema->find_field(30)->field_type()->type_id(), TypeID::VARIANT);
+}
+
//TEST(SchemaParserTest, parse_invalid_json) {
// EXPECT_THROW(SchemaParser::from_json(invalid_json), doris::Exception);
//}
diff --git a/be/test/format/table/iceberg/type_test.cpp b/be/test/format/table/iceberg/type_test.cpp
index 73bd2fd54df..8ede6af24b7 100644
--- a/be/test/format/table/iceberg/type_test.cpp
+++ b/be/test/format/table/iceberg/type_test.cpp
@@ -40,6 +40,12 @@ TEST(BinaryTypeTest, to_string) {
EXPECT_EQ(binary_type.type_id(), TypeID::BINARY);
}
+TEST(VariantTypeTest, to_string) {
+ VariantType variant_type;
+ EXPECT_EQ(variant_type.to_string(), "variant");
+ EXPECT_EQ(variant_type.type_id(), TypeID::VARIANT);
+}
+
TEST(FixedTypeTest, to_string) {
FixedType fixed_type(16);
EXPECT_EQ(fixed_type.to_string(), "fixed[16]");
@@ -146,6 +152,7 @@ TEST(PrimitiveTypesTest, from_primitive_string) {
EXPECT_EQ(Types::from_primitive_string("string")->type_id(), TypeID::STRING);
EXPECT_EQ(Types::from_primitive_string("uuid")->type_id(), TypeID::UUID);
EXPECT_EQ(Types::from_primitive_string("binary")->type_id(), TypeID::BINARY);
+ EXPECT_EQ(Types::from_primitive_string("variant")->type_id(), TypeID::VARIANT);
EXPECT_EQ(Types::from_primitive_string("fixed[16]")->type_id(), TypeID::FIXED);
EXPECT_EQ(Types::from_primitive_string("decimal(10, 2)")->type_id(), TypeID::DECIMAL);
}
diff --git a/be/test/format/transformer/vparquet_writer_test.cpp b/be/test/format/transformer/vparquet_writer_test.cpp
index d296f2c586d..c4b05cf045e 100644
--- a/be/test/format/transformer/vparquet_writer_test.cpp
+++ b/be/test/format/transformer/vparquet_writer_test.cpp
@@ -32,14 +32,17 @@
#include "core/column/column_array.h"
#include "core/column/column_decimal.h"
#include "core/column/column_nullable.h"
+#include "core/column/variant_v2/column_variant_v2.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_timestamp_ns.h"
+#include "core/data_type/data_type_variant_v2.h"
#include "exec/sink/writer/iceberg/viceberg_partition_writer.h"
#include "exec/sink/writer/vhive_partition_writer.h"
+#include "exprs/function/parse/variant_string_parse.h"
#include "format/table/iceberg/schema_parser.h"
#include "format/transformer/viceberg_parquet_writer.h"
#include "io/fs/file_writer.h"
@@ -166,6 +169,164 @@ TEST_F(VParquetWriterTest, WritesIcebergPrimitiveAndCollectsMetrics) {
EXPECT_EQ(std::string("\x03\x00\x00\x00", 4), stats.upper_bounds.at(7));
}
+TEST_F(VParquetWriterTest, WritesIcebergVariantAndCollectsLogicalMetrics) {
+ auto variant_type = std::make_shared<DataTypeVariantV2>();
+ auto nullable_variant_type = make_nullable(variant_type);
+ VExprContextSPtrs output_exprs =
+ MockSlotRef::create_mock_contexts(DataTypes {nullable_variant_type});
+
+ const std::string schema_json = R"({
+ "type": "struct",
+ "fields": [
+ {"id": 2, "name": "payload", "required": false, "type": "variant"}
+ ]
+ })";
+ std::unique_ptr<iceberg::Schema> schema = iceberg::SchemaParser::from_json(schema_json);
+
+ io::FileWriterPtr file_writer;
+ ASSERT_TRUE(_fs->create_file(_file_path, &file_writer).ok());
+ RuntimeState state;
+ state.set_timezone("UTC");
+ ParquetFileOptions options {.compression_type = TParquetCompressionType::UNCOMPRESSED,
+ .parquet_version = TParquetVersion::PARQUET_1_0,
+ .parquet_disable_dictionary = false,
+ .enable_int96_timestamps = false};
+ VIcebergParquetWriter transformer(&state, file_writer.get(), output_exprs, {"payload"}, false,
+ options, &schema_json, *schema);
+ ASSERT_TRUE(transformer.open().ok());
+
+ JsonStringToVariantEncoder encoder({.max_json_key_length = 1024,
+ .throw_on_invalid_json = true,
+ .check_duplicate_json_path = false});
+ for (std::string_view json : {std::string_view {R"({"name":"doris","n":1})"},
+ std::string_view {"null"}, std::string_view {R"([1,2,3])"}}) {
+ encoder.add_json({json.data(), json.size()});
+ }
+ auto variant_column = ColumnVariantV2::create();
+ variant_column->insert_encoded_batch(encoder.finish_batch());
+ auto null_map = ColumnUInt8::create();
+ null_map->get_data().assign({0, 0, 1});
+ auto nullable_column = ColumnNullable::create(std::move(variant_column), std::move(null_map));
+
+ Block block;
+ block.insert(
+ ColumnWithTypeAndName(std::move(nullable_column), nullable_variant_type, "payload"));
+ ASSERT_TRUE(transformer.write(block).ok());
+ ASSERT_TRUE(transformer.close().ok());
+
+ TIcebergColumnStats stats;
+ ASSERT_TRUE(transformer.collect_file_statistics_after_close(&stats).ok());
+ EXPECT_EQ(0, stats.column_sizes.count(-1));
+ EXPECT_EQ(0, stats.value_counts.count(-1));
+ EXPECT_EQ(0, stats.null_value_counts.count(-1));
+ EXPECT_EQ(0, stats.column_sizes.count(2));
+ ASSERT_EQ(1, stats.value_counts.count(2));
+ EXPECT_EQ(3, stats.value_counts.at(2));
+ ASSERT_TRUE(stats.__isset.null_value_counts);
+ ASSERT_EQ(1, stats.null_value_counts.count(2));
+ EXPECT_EQ(1, stats.null_value_counts.at(2));
+ if (stats.__isset.lower_bounds) {
+ EXPECT_EQ(0, stats.lower_bounds.count(2));
+ }
+ if (stats.__isset.upper_bounds) {
+ EXPECT_EQ(0, stats.upper_bounds.count(2));
+ }
+
+ auto reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ const auto* root = reader->metadata()->schema()->group_node();
+ ASSERT_EQ(1, root->field_count());
+ const auto& payload = root->field(0);
+ ASSERT_NE(nullptr, payload->logical_type());
+ EXPECT_TRUE(payload->logical_type()->is_variant());
+ EXPECT_EQ(2, payload->field_id());
+ const auto& payload_group = static_cast<const ::parquet::schema::GroupNode&>(*payload);
+ ASSERT_EQ(2, payload_group.field_count());
+ EXPECT_EQ(-1, payload_group.field(0)->field_id());
+ EXPECT_EQ(-1, payload_group.field(1)->field_id());
+}
+
+TEST_F(VParquetWriterTest, WritesNestedIcebergVariant) {
+ auto variant_type = std::make_shared<DataTypeVariantV2>();
+ auto array_type = std::make_shared<DataTypeArray>(variant_type);
+ auto nullable_array_type = make_nullable(array_type);
+ VExprContextSPtrs output_exprs =
+ MockSlotRef::create_mock_contexts(DataTypes {nullable_array_type});
+
+ const std::string schema_json = R"({
+ "type": "struct",
+ "fields": [
+ {
+ "id": 2,
+ "name": "events",
+ "required": false,
+ "type": {
+ "type": "list",
+ "element-id": 3,
+ "element": "variant",
+ "element-required": false
+ }
+ }
+ ]
+ })";
+ std::unique_ptr<iceberg::Schema> schema = iceberg::SchemaParser::from_json(schema_json);
+
+ io::FileWriterPtr file_writer;
+ ASSERT_TRUE(_fs->create_file(_file_path, &file_writer).ok());
+ RuntimeState state;
+ state.set_timezone("UTC");
+ ParquetFileOptions options {.compression_type = TParquetCompressionType::UNCOMPRESSED,
+ .parquet_version = TParquetVersion::PARQUET_1_0,
+ .parquet_disable_dictionary = false,
+ .enable_int96_timestamps = false};
+ VIcebergParquetWriter transformer(&state, file_writer.get(), output_exprs, {"events"}, false,
+ options, &schema_json, *schema);
+ ASSERT_TRUE(transformer.open().ok());
+
+ JsonStringToVariantEncoder encoder({.max_json_key_length = 1024,
+ .throw_on_invalid_json = true,
+ .check_duplicate_json_path = false});
+ for (std::string_view json :
+ {std::string_view {R"({"kind":"nested"})"}, std::string_view {"null"},
+ std::string_view {"{}"}, std::string_view {"7"}}) {
+ encoder.add_json({json.data(), json.size()});
+ }
+ auto variant_values = ColumnVariantV2::create();
+ variant_values->insert_encoded_batch(encoder.finish_batch());
+ auto element_nulls = ColumnUInt8::create();
+ element_nulls->get_data().assign({0, 0, 1, 0});
+ auto nullable_values =
+ ColumnNullable::create(std::move(variant_values), std::move(element_nulls));
+ auto offsets = ColumnArray::ColumnOffsets::create();
+ offsets->get_data().assign({3, 4});
+ auto events = ColumnArray::create(std::move(nullable_values), std::move(offsets));
+ auto outer_nulls = ColumnUInt8::create();
+ outer_nulls->get_data().assign({0, 1});
+ auto nullable_events = ColumnNullable::create(std::move(events), std::move(outer_nulls));
+
+ Block block;
+ block.insert(ColumnWithTypeAndName(std::move(nullable_events), nullable_array_type, "events"));
+ ASSERT_TRUE(transformer.write(block).ok());
+ ASSERT_TRUE(transformer.close().ok());
+
+ TIcebergColumnStats stats;
+ ASSERT_TRUE(transformer.collect_file_statistics_after_close(&stats).ok());
+ EXPECT_EQ(0, stats.value_counts.count(-1));
+ EXPECT_EQ(0, stats.null_value_counts.count(-1));
+ ASSERT_EQ(1, stats.value_counts.count(3));
+ EXPECT_EQ(0, stats.lower_bounds.count(3));
+ EXPECT_EQ(0, stats.upper_bounds.count(3));
+
+ auto reader = ::parquet::ParquetFileReader::OpenFile(_file_path, false);
+ const auto* root = reader->metadata()->schema()->group_node();
+ const auto& events_group = static_cast<const ::parquet::schema::GroupNode&>(*root->field(0));
+ const auto& list_group =
+ static_cast<const ::parquet::schema::GroupNode&>(*events_group.field(0));
+ const auto& element = list_group.field(0);
+ ASSERT_NE(nullptr, element->logical_type());
+ EXPECT_TRUE(element->logical_type()->is_variant());
+ EXPECT_EQ(3, element->field_id());
+}
+
TEST_F(VParquetWriterTest, IcebergDecimalBoundsUseMinimalSignedBytes) {
for (const auto& [unscaled, expected] :
std::vector<std::pair<int64_t, std::string>> {{0, std::string("\x00", 1)},
diff --git a/be/test/format_v2/table/iceberg_reader_test.cpp b/be/test/format_v2/table/iceberg_reader_test.cpp
index 09e103626ba..9b70c0ada25 100644
--- a/be/test/format_v2/table/iceberg_reader_test.cpp
+++ b/be/test/format_v2/table/iceberg_reader_test.cpp
@@ -1893,6 +1893,108 @@ TEST(IcebergV2ReaderTest, PreparesIcebergNonFiniteInitialDefaults) {
}
}
+TEST(IcebergV2ReaderTest, AnnotateVariantInitialDefaultMustBeNull) {
+ const auto variant_type = make_nullable(std::make_shared<DataTypeVariantV2>());
+
+ auto optional_variant =
+ external_schema_field("payload", 1, {}, std::nullopt,
+ external_primitive_type(TPrimitiveType::VARIANT), false, true);
+ TFileScanRangeParams valid_scan_params;
+ valid_scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ valid_scan_params.__set_current_schema_id(100);
+ valid_scan_params.__set_history_schema_info(
+ {external_schema(100, {std::move(optional_variant)})});
+
+ ColumnDefinition valid_column;
+ valid_column.name = "payload";
+ valid_column.type = variant_type;
+ ProjectedColumnBuildContext valid_context {.scan_params = &valid_scan_params};
+ doris::format::iceberg::IcebergTableReader reader;
+ ASSERT_TRUE(reader.annotate_projected_column(TFileScanSlotInfo(), &valid_context, &valid_column)
+ .ok());
+ ASSERT_TRUE(valid_column.is_optional.has_value());
+ EXPECT_TRUE(*valid_column.is_optional);
+ EXPECT_FALSE(valid_column.initial_default_value.has_value());
+ EXPECT_EQ(valid_column.default_expr, nullptr);
+ ASSERT_TRUE(valid_context.schema_column.has_value());
+ EXPECT_TRUE(valid_context.schema_column->type->equals(*variant_type));
+
+ auto malformed_variant =
+ external_schema_field("payload", 1, {}, "{\"source\":\"malformed\"}",
+ external_primitive_type(TPrimitiveType::VARIANT), false, true);
+ TFileScanRangeParams malformed_scan_params;
+ malformed_scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ malformed_scan_params.__set_current_schema_id(101);
+ malformed_scan_params.__set_history_schema_info(
+ {external_schema(101, {std::move(malformed_variant)})});
+
+ ColumnDefinition malformed_column;
+ malformed_column.name = "payload";
+ malformed_column.type = variant_type;
+ ProjectedColumnBuildContext malformed_context {.scan_params = &malformed_scan_params};
+ const auto status = reader.annotate_projected_column(TFileScanSlotInfo(), &malformed_context,
+ &malformed_column);
+ EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status;
+ EXPECT_NE(status.to_string().find("VARIANT initial-default for field 'payload' must be NULL"),
+ std::string::npos)
+ << status;
+}
+
+TEST(IcebergV2ReaderTest, StructInitialDefaultKeepsVariantChildNull) {
+ const auto variant_type = make_nullable(std::make_shared<DataTypeVariantV2>());
+ const auto struct_type = make_nullable(
+ std::make_shared<DataTypeStruct>(DataTypes {variant_type}, Strings {"payload"}));
+
+ auto optional_variant =
+ external_schema_field("payload", 2, {}, std::nullopt,
+ external_primitive_type(TPrimitiveType::VARIANT), false, true);
+ auto valid_struct =
+ external_struct_schema_field("event", 1, {std::move(optional_variant)}, true, "{}");
+ TFileScanRangeParams valid_scan_params;
+ valid_scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ valid_scan_params.__set_current_schema_id(100);
+ valid_scan_params.__set_history_schema_info({external_schema(100, {std::move(valid_struct)})});
+
+ ColumnDefinition valid_column;
+ valid_column.name = "event";
+ valid_column.type = struct_type;
+ ProjectedColumnBuildContext valid_context {.scan_params = &valid_scan_params};
+ doris::format::iceberg::IcebergTableReader reader;
+ ASSERT_TRUE(reader.annotate_projected_column(TFileScanSlotInfo(), &valid_context, &valid_column)
+ .ok());
+ ASSERT_NE(valid_column.default_expr, nullptr);
+ const auto* literal = dynamic_cast<const VLiteral*>(valid_column.default_expr->root().get());
+ ASSERT_NE(literal, nullptr);
+ EXPECT_TRUE(literal->get_data_type()->equals(*struct_type));
+ Field value;
+ literal->get_column_ptr()->get(0, value);
+ const auto& struct_fields = value.get<TYPE_STRUCT>();
+ ASSERT_EQ(struct_fields.size(), 1);
+ EXPECT_TRUE(struct_fields[0].is_null());
+
+ auto malformed_variant =
+ external_schema_field("payload", 2, {}, std::nullopt,
+ external_primitive_type(TPrimitiveType::VARIANT), false, true);
+ auto malformed_struct = external_struct_schema_field(
+ "event", 1, {std::move(malformed_variant)}, true, "{\"2\":{\"source\":\"malformed\"}}");
+ TFileScanRangeParams malformed_scan_params;
+ malformed_scan_params.__set_iceberg_scan_semantics_version(ICEBERG_SCAN_SEMANTICS_VERSION_2);
+ malformed_scan_params.__set_current_schema_id(101);
+ malformed_scan_params.__set_history_schema_info(
+ {external_schema(101, {std::move(malformed_struct)})});
+
+ ColumnDefinition malformed_column;
+ malformed_column.name = "event";
+ malformed_column.type = struct_type;
+ ProjectedColumnBuildContext malformed_context {.scan_params = &malformed_scan_params};
+ const auto status = reader.annotate_projected_column(TFileScanSlotInfo(), &malformed_context,
+ &malformed_column);
+ EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status;
+ EXPECT_NE(status.to_string().find("VARIANT initial-default for field 'payload' must be NULL"),
+ std::string::npos)
+ << status;
+}
+
TEST(IcebergV2ReaderTest, AnnotateBuildsComplexInitialDefaults) {
const auto required_int_type = std::make_shared<DataTypeInt32>();
const auto optional_string_type = make_nullable(std::make_shared<DataTypeString>());
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
index 19776b91542..8659fc77340 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogOps.java
@@ -512,6 +512,9 @@ public interface IcebergCatalogOps {
public void addColumn(String dbName, String tableName, IcebergColumnChange column,
ConnectorColumnPosition position) {
withTable(dbName, tableName, table -> {
+ if (column.getType().isVariantType()) {
+ IcebergWriterHelper.validateVariantColumn(table, column.getName());
+ }
IcebergNestedColumnEvolution.validateNoCaseInsensitiveSiblingCollision(
table.schema().asStruct(), "", column.getName(), null, "add");
UpdateSchema updateSchema = table.updateSchema();
@@ -526,6 +529,11 @@ public interface IcebergCatalogOps {
@Override
public void addColumns(String dbName, String tableName, List<IcebergColumnChange> columns) {
withTable(dbName, tableName, table -> {
+ for (IcebergColumnChange column : columns) {
+ if (column.getType().isVariantType()) {
+ IcebergWriterHelper.validateVariantColumn(table, column.getName());
+ }
+ }
IcebergNestedColumnEvolution.validateNoCaseInsensitiveTopLevelCollisions(table.schema(), columns);
UpdateSchema updateSchema = table.updateSchema();
for (IcebergColumnChange column : columns) {
@@ -577,7 +585,17 @@ public interface IcebergCatalogOps {
// #65329 omit-preserves: an omitted COMMENT keeps the field's current doc rather than clearing it
// (mirror of IcebergNestedColumnEvolution.modifyColumn / legacy IcebergMetadataOps).
String targetComment = commentSpecified ? column.getComment() : current.doc();
- if (newType.isPrimitiveType()) {
+ if (current.type().isVariantType() != newType.isVariantType()) {
+ throw new DorisConnectorException("Cannot change column type involving Iceberg VARIANT: "
+ + column.getName() + ": " + current.type() + " -> " + newType);
+ }
+ if (newType.isVariantType()) {
+ // A VARIANT modify changes only metadata, so it does not check the file format: an existing
+ // v3 ORC table can hold a VARIANT column although Doris writes VARIANT values only to Parquet.
+ if (!Objects.equals(current.doc(), targetComment)) {
+ updateSchema.updateColumnDoc(currentName, targetComment);
+ }
+ } else if (newType.isPrimitiveType()) {
// Reject a complex -> primitive change with a clean message before iceberg's updateColumn leaks
// a raw type-diff error — parity with IcebergNestedColumnEvolution.modifyColumn / legacy
// IcebergMetadataOps (symmetric with the non-complex -> complex guard below).
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index a2f63a69239..134bf0cd9de 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -1066,6 +1066,11 @@ public class IcebergConnectorMetadata implements ConnectorMetadata {
// CREATE properties (ConnectorFactory.createConnector(catalogProperty.getProperties())).
Map<String, String> tableProperties =
IcebergSchemaBuilder.buildTableProperties(request.getProperties(), properties);
+ if (request.getColumns().stream().anyMatch(column -> IcebergSchemaBuilder.isVariant(column.getType()))) {
+ IcebergWriterHelper.validateVariantWriteFormat(
+ IcebergSchemaBuilder.getEffectiveFormatVersion(tableProperties, properties),
+ IcebergSchemaBuilder.getEffectiveFileFormat(tableProperties, properties));
+ }
applyCreateTableComment(tableProperties, request.getComment());
try {
executeAuthenticated(() -> {
@@ -1486,7 +1491,7 @@ public class IcebergConnectorMetadata implements ConnectorMetadata {
if (!column.isNullable()) {
throw new DorisConnectorException("New nested field '" + path.getFullPath() + "' must be nullable");
}
- Type icebergType = IcebergSchemaBuilder.buildColumnType(column.getType());
+ Type icebergType = IcebergSchemaBuilder.buildNestedFieldType(column.getType());
IcebergColumnChange change = new IcebergColumnChange(path.getLeafName(), icebergType,
column.getComment(), null, column.isNullable());
try {
@@ -1573,7 +1578,7 @@ public class IcebergConnectorMetadata implements ConnectorMetadata {
}
Type icebergType;
try {
- icebergType = IcebergSchemaBuilder.buildColumnType(column.getType());
+ icebergType = IcebergSchemaBuilder.buildNestedFieldType(column.getType());
} catch (DorisConnectorException buildError) {
// Preserve the complete target identity so error parity cannot bind a same-named top-level field.
throw upgradeNestedModifyError(iceHandle, path, column, buildError);
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilder.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilder.java
index c5860163cf3..c6026ef1f01 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilder.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilder.java
@@ -25,6 +25,7 @@ import org.apache.doris.connector.spi.ddl.ConnectorPartitionSpec;
import org.apache.doris.connector.spi.ddl.ConnectorSortField;
import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.NullOrder;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.RowLevelOperationMode;
@@ -71,6 +72,9 @@ import java.util.UUID;
public final class IcebergSchemaBuilder {
private static final List<String> ICEBERG_TABLE_LOCATION_CHILD_DIRS = java.util.Arrays.asList("data", "metadata");
+ // The Flink and Spark nickname of write.format.default, which wins over it (as IcebergWriterHelper reads them).
+ private static final String WRITE_FORMAT = "write-format";
+ static final String NESTED_VARIANT_DDL_ERROR = "Iceberg VARIANT DDL currently supports only top-level columns";
private IcebergSchemaBuilder() {
}
@@ -91,7 +95,7 @@ public final class IcebergSchemaBuilder {
List<Types.NestedField> fields = new ArrayList<>(columns.size());
for (int i = 0; i < columns.size(); i++) {
ConnectorColumn col = columns.get(i);
- Type type = convert(col.getType(), ids);
+ Type type = convertColumn(col.getType(), ids);
if (col.isNullable()) {
fields.add(Types.NestedField.optional(i, col.getName(), type, col.getComment()));
} else {
@@ -113,6 +117,9 @@ public final class IcebergSchemaBuilder {
private static Type convert(ConnectorType type, IdAllocator ids) {
String name = type.getTypeName().toUpperCase(Locale.ROOT);
switch (name) {
+ case "VARIANT":
+ case "VARIANT_COMPUTE_V2":
+ throw new DorisConnectorException(NESTED_VARIANT_DDL_ERROR);
case "ARRAY": {
// Element type/ids first, then the list's element id (post-order, matching legacy visitor).
Type element = convert(type.getChildren().get(0), ids);
@@ -150,6 +157,17 @@ public final class IcebergSchemaBuilder {
}
}
+ /** Converts the type of a top-level column, the only place a VARIANT may be declared. */
+ private static Type convertColumn(ConnectorType type, IdAllocator ids) {
+ return isVariant(type) ? Types.VariantType.get() : convert(type, ids);
+ }
+
+ /** Whether {@code type} is a VARIANT, as declared by DDL or read from an Iceberg table. */
+ static boolean isVariant(ConnectorType type) {
+ String name = type.getTypeName().toUpperCase(Locale.ROOT);
+ return "VARIANT".equals(name) || "VARIANT_COMPUTE_V2".equals(name);
+ }
+
/**
* Builds the Iceberg {@link PartitionSpec} against {@code schema} from the neutral partition spec.
* String-driven port of {@code IcebergUtils.solveIcebergPartitionSpec}: each field's transform name +
@@ -320,6 +338,33 @@ public final class IcebergSchemaBuilder {
}
}
+ /**
+ * The effective iceberg data file format for a CREATE TABLE: catalog {@code table-default} properties, then
+ * the table request, then catalog {@code table-override} properties, each overriding the former; of the
+ * resulting {@code write-format} nickname and standard {@code write.format.default}, the nickname wins, as
+ * for an existing table ({@link IcebergWriterHelper#getFileFormat}). Without either, Parquet.
+ */
+ public static FileFormat getEffectiveFileFormat(Map<String, String> requestProperties,
+ Map<String, String> catalogProperties) {
+ Map<String, String> effective = new HashMap<>();
+ copyFileFormat(effective, catalogProperties, CatalogProperties.TABLE_DEFAULT_PREFIX);
+ copyFileFormat(effective, requestProperties, "");
+ copyFileFormat(effective, catalogProperties, CatalogProperties.TABLE_OVERRIDE_PREFIX);
+ String fileFormat = effective.containsKey(WRITE_FORMAT) ? effective.get(WRITE_FORMAT)
+ : effective.getOrDefault(TableProperties.DEFAULT_FILE_FORMAT, FileFormat.PARQUET.name());
+ return IcebergWriterHelper.toFileFormat(fileFormat);
+ }
+
+ private static void copyFileFormat(Map<String, String> effective, Map<String, String> properties,
+ String prefix) {
+ for (String property : new String[] {WRITE_FORMAT, TableProperties.DEFAULT_FILE_FORMAT}) {
+ String value = properties.get(prefix + property);
+ if (value != null) {
+ effective.put(property, value);
+ }
+ }
+ }
+
/**
* Builds the iceberg {@link Type} for a SINGLE column added/modified on an EXISTING table, reusing the
* same neutral-type conversion as {@link #buildSchema} (scalars via {@link IcebergTypeMapping}, plus
@@ -333,6 +378,15 @@ public final class IcebergSchemaBuilder {
* @throws DorisConnectorException if the column type cannot be represented in iceberg
*/
public static Type buildColumnType(ConnectorType type) {
+ return convertColumn(type, new IdAllocator(0));
+ }
+
+ /**
+ * Like {@link #buildColumnType}, for a field nested in a STRUCT, which cannot be a VARIANT.
+ *
+ * @throws DorisConnectorException if the field type cannot be represented in iceberg
+ */
+ public static Type buildNestedFieldType(ConnectorType type) {
return convert(type, new IdAllocator(0));
}
@@ -408,6 +462,8 @@ public final class IcebergSchemaBuilder {
} catch (NumberFormatException e) {
throw new IllegalArgumentException("Invalid Decimal string: " + value, e);
}
+ case VARIANT:
+ throw new IllegalArgumentException("Iceberg VARIANT default values must be NULL");
default:
throw new IllegalArgumentException("Cannot parse unknown type: " + type);
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
index 1e8600115b7..0ea68a006c7 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergSchemaUtils.java
@@ -521,6 +521,9 @@ public final class IcebergSchemaUtils {
}
static String serializeInitialDefault(Type type, Object value, boolean enableTimestampTz) {
+ if (type.typeId() == TypeID.VARIANT) {
+ throw new DorisConnectorException("Iceberg VARIANT initial-default must be NULL");
+ }
if (type.isNestedType()) {
return SingleValueParser.toJson(type, value);
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
index d929c5938cd..57f2fa1e338 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
@@ -66,6 +66,7 @@ import org.apache.iceberg.TableProperties;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types.NestedField;
import org.apache.iceberg.util.LocationUtil;
+import org.apache.iceberg.util.PropertyUtil;
import java.util.ArrayList;
import java.util.Arrays;
@@ -117,6 +118,8 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider {
// Older backends cannot hash a BINARY or FIXED value (ColumnVarbinary) in the exchange.
private static final int SUPPORT_BINARY_ROUTING_EXEC_VERSION = 19;
private static final String TRANSFORM_PARTITION_FUNCTION = "iceberg_transform";
+ // Shredded VARIANT values are written as typed Parquet columns, which Doris does not produce.
+ private static final String PARQUET_SHRED_VARIANTS = "write.parquet.shred-variants";
// Legacy IcebergUtils compression-codec property keys (connector-local copies; iceberg SDK has no
// constant for the doris/spark-sql forms).
@@ -194,7 +197,6 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider {
@Override
public ConnectorSinkPlan planWrite(ConnectorSession session, ConnectorWriteHandle handle) {
- validateWriteSchema(handle);
IcebergTableHandle tableHandle = (IcebergTableHandle) handle.getTableHandle();
IcebergConnectorTransaction transaction = currentTransaction(session);
@@ -204,6 +206,10 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider {
// transaction switches on at commit time (Append vs ReplacePartitions / OverwriteFiles).
IcebergWriteSchemaContext schemaContext = handle.getWriteOperation() == WriteOperation.REWRITE
? null : resolveWriteSchema(session, tableHandle, handle.getBranchName());
+ if (schemaContext != null) {
+ // REWRITE resolves its schema after beginWrite and checks it in buildRewriteSink.
+ validateVariantWrite(handle, schemaContext);
+ }
validateNestedPartitionWriteCompatibility(handle, schemaContext);
IcebergWriteContext writeContext = buildWriteContext(handle, schemaContext);
transaction.beginWrite(session, tableHandle.getDbName(), tableHandle.getTableName(), writeContext);
@@ -392,22 +398,28 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider {
|| "TIMESTAMPTZ".equals(typeName);
}
- static void validateWriteSchema(List<ConnectorColumn> columns, boolean writesDataFiles) {
- if (!writesDataFiles) {
+ /**
+ * Checks a write of data files whose target holds a VARIANT, at any depth, against the table it writes:
+ * VARIANT needs format-version 3, and Doris writes it only unshredded to Parquet data files.
+ */
+ static void validateVariantWrite(List<ConnectorColumn> columns, boolean writesDataFiles, int formatVersion,
+ FileFormat fileFormat, Map<String, String> writerProperties) {
+ if (!writesDataFiles || columns.stream().noneMatch(column -> containsVariant(column.getType()))) {
return;
}
- if (columns.stream().anyMatch(column -> containsVariant(column.getType()))) {
- // Reject the whole data-file write: validating only selected columns would let an
- // unchanged Variant target flow through a writer that cannot preserve its physical identity.
- throw new DorisConnectorException(
- "Iceberg VARIANT columns are read-only and cannot be written");
+ IcebergWriterHelper.validateVariantWriteFormat(formatVersion, fileFormat);
+ if (PropertyUtil.propertyAsBoolean(writerProperties, PARQUET_SHRED_VARIANTS, false)) {
+ throw new DorisConnectorException("Doris currently supports only unshredded Iceberg VARIANT writes; set "
+ + PARQUET_SHRED_VARIANTS + "=false before writing");
}
}
- static void validateWriteSchema(ConnectorWriteHandle handle) {
+ private static void validateVariantWrite(ConnectorWriteHandle handle, IcebergWriteSchemaContext schemaContext) {
// The explicit INSERT column list can omit an unchanged Variant column, but the data writer
// still binds the complete target schema and therefore must validate that complete shape.
- validateWriteSchema(handle.getBoundTargetColumns(), handle.isWritesDataFiles());
+ validateVariantWrite(handle.getBoundTargetColumns(), handle.isWritesDataFiles(),
+ schemaContext.getFormatVersion(), schemaContext.getFileFormat(),
+ schemaContext.getWriterProperties());
}
private static void validateNestedPartitionWriteCompatibility(
@@ -895,8 +907,9 @@ public class IcebergWritePlanProvider implements ConnectorWritePlanProvider {
IcebergWriteSchemaContext rewriteSchemaContext = IcebergWriteSchemaContext.create(
table, tableHandle.getTableName(), Optional.empty(),
mappingVarbinaryEnabled(), mappingTimestampTzEnabled());
- // REWRITE resolves its schema after beginWrite, so its rolling-upgrade fence must run here
- // against the same partition spec that the compaction sink will actually use.
+ // REWRITE resolves its schema after beginWrite, so its Variant check and rolling-upgrade fence must run
+ // here against the same schema and partition spec that the compaction sink will actually use.
+ validateVariantWrite(handle, rewriteSchemaContext);
validateNestedPartitionWriteCompatibility(handle, rewriteSchemaContext);
TIcebergTableSink tSink = buildSink(table, tableHandle, handle, rewriteSchemaContext);
tSink.setWriteType(TIcebergWriteType.REWRITE);
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
index 12c592c8fa1..ad535befa34 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
@@ -474,6 +474,8 @@ final class IcebergWriteSchemaContext {
enableMappingVarbinary, enableMappingTimestampTz));
}
return "named_struct(" + String.join(", ", fields) + ")";
+ case VARIANT:
+ throw new DorisConnectorException("Iceberg VARIANT write-default must be NULL");
default:
throw new DorisConnectorException("Unsupported Iceberg write-default type: " + type);
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
index a10a95cb45b..29c9ba68fae 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriterHelper.java
@@ -17,6 +17,7 @@
package org.apache.doris.connector.iceberg;
+import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.thrift.TFileContent;
import org.apache.doris.thrift.TIcebergColumnStats;
import org.apache.doris.thrift.TIcebergCommitData;
@@ -89,10 +90,38 @@ final class IcebergWriterHelper {
// Local copy (value 3), mirroring the connector's per-class private ICEBERG_ROW_LINEAGE_MIN_VERSION in
// IcebergConnectorMetadata; the self-contained helper does not import fe-core IcebergUtils.
private static final int ICEBERG_ROW_LINEAGE_MIN_VERSION = 3;
+ // Iceberg introduced the VARIANT type in format-version 3.
+ private static final int ICEBERG_VARIANT_MIN_VERSION = 3;
private IcebergWriterHelper() {
}
+ /**
+ * Checks the format of a table that is to hold VARIANT values Doris writes: VARIANT needs format-version 3,
+ * and Doris writes it only to Parquet data files.
+ */
+ static void validateVariantWriteFormat(int formatVersion, FileFormat fileFormat) {
+ if (formatVersion < ICEBERG_VARIANT_MIN_VERSION) {
+ throw new DorisConnectorException(
+ "Iceberg VARIANT writes require table format-version 3, but found " + formatVersion);
+ }
+ if (fileFormat != FileFormat.PARQUET) {
+ throw new DorisConnectorException(
+ "Iceberg VARIANT writes require Parquet data files, but found " + fileFormat);
+ }
+ }
+
+ /** Checks that {@code table} can take the new VARIANT column {@code columnName} that Doris will write. */
+ static void validateVariantColumn(Table table, String columnName) {
+ if (getFormatVersion(table) < ICEBERG_VARIANT_MIN_VERSION) {
+ throw new DorisConnectorException(
+ "Iceberg VARIANT column " + columnName + " requires table format-version 3");
+ }
+ if (getFileFormat(table) != FileFormat.PARQUET) {
+ throw new DorisConnectorException("Iceberg VARIANT column " + columnName + " requires Parquet data files");
+ }
+ }
+
/**
* Converts the BE data-file commit fragments into an iceberg {@link WriteResult} of {@link DataFile}s
* (the INSERT / OVERWRITE / MERGE-data path). A partitioned table requires non-empty partition values per
@@ -413,7 +442,7 @@ final class IcebergWriterHelper {
return toFileFormat(resolveFileFormatName(table, table.properties(), id, cache));
}
- private static FileFormat toFileFormat(String fileFormatName) {
+ static FileFormat toFileFormat(String fileFormatName) {
if (fileFormatName.toLowerCase().contains(ORC_NAME)) {
return FileFormat.ORC;
} else if (fileFormatName.toLowerCase().contains(PARQUET_NAME)) {
@@ -560,8 +589,9 @@ final class IcebergWriterHelper {
.anyMatch(field -> MetricsUtil.metricsMode(writerSchema, metricsConfig, field.fieldId())
!= MetricsModes.None.get());
}
+ // A Parquet VARIANT reports the value and null counts of its own field id.
return TypeUtil.indexById(writerSchema.asStruct()).values().stream()
- .filter(field -> field.type().isPrimitiveType())
+ .filter(field -> field.type().isPrimitiveType() || field.type().isVariantType())
.anyMatch(field -> MetricsUtil.metricsMode(writerSchema, metricsConfig, field.fieldId())
!= MetricsModes.None.get());
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
index e23b0cfb309..9ae6be015b2 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/CatalogBackedIcebergCatalogOpsColumnEvolutionTest.java
@@ -38,7 +38,9 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
/**
@@ -333,6 +335,80 @@ public class CatalogBackedIcebergCatalogOpsColumnEvolutionTest {
Assertions.assertTrue(ex.getMessage().contains("does not exist"));
}
+ // ---------- VARIANT ----------
+
+ /** Creates {@code db1.<name>} with an id and, when {@code withVariant}, a required VARIANT payload. */
+ private void createTable(String name, String formatVersion, String fileFormat, boolean withVariant) {
+ Map<String, String> props = new HashMap<>(IcebergSchemaBuilder.buildTableProperties(Collections.emptyMap()));
+ props.put("format-version", formatVersion);
+ props.put("write.format.default", fileFormat);
+ Schema schema = withVariant
+ ? new Schema(Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.required(2, "payload", Types.VariantType.get(), "old"))
+ : new Schema(Types.NestedField.optional(1, "id", Types.LongType.get()));
+ ops.createTable("db1", name, schema, PartitionSpec.unpartitioned(), null, props);
+ }
+
+ @Test
+ public void testAddVariantColumnNeedsFormatThreeParquetTable() {
+ // Doris writes VARIANT values only to format-version 3 Parquet data files, so a VARIANT column is added
+ // only to such a table, alone or with other columns.
+ createTable("v2_parquet", "2", "parquet", false);
+ createTable("v3_orc", "3", "orc", false);
+ createTable("v3_parquet", "3", "parquet", false);
+ IcebergColumnChange payload = change("payload", Types.VariantType.get(), "c", true);
+ DorisConnectorException v2 = Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.addColumn("db1", "v2_parquet", payload, null));
+ Assertions.assertEquals("Iceberg VARIANT column payload requires table format-version 3", v2.getMessage());
+ DorisConnectorException orc = Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.addColumns("db1", "v3_orc", Arrays.asList(
+ change("a", Types.IntegerType.get(), "c", true), payload)));
+ Assertions.assertEquals("Iceberg VARIANT column payload requires Parquet data files", orc.getMessage());
+ Assertions.assertNull(ops.loadTable("db1", "v3_orc").schema().findField("a"));
+
+ ops.addColumn("db1", "v3_parquet", payload, null);
+ ops.addColumns("db1", "v3_parquet", Arrays.asList(
+ change("a", Types.IntegerType.get(), "c", true),
+ change("other_payload", Types.VariantType.get(), "c", true)));
+ Schema schema = ops.loadTable("db1", "v3_parquet").schema();
+ Assertions.assertEquals(Types.VariantType.get(), schema.findField("payload").type());
+ Assertions.assertEquals(Types.VariantType.get(), schema.findField("other_payload").type());
+ }
+
+ @Test
+ public void testModifyVariantColumnChangesOnlyItsMetadata() {
+ // A VARIANT modify only changes metadata, so it skips the file format check: a v3 ORC table can hold a
+ // VARIANT column that another engine wrote. The comment changes, the column becomes nullable and moves,
+ // and its type stays VARIANT.
+ createTable("orc_variant", "3", "orc", true);
+ ops.modifyColumn("db1", "orc_variant", change("payload", Types.VariantType.get(), "new doc", true), true,
+ ConnectorColumnPosition.FIRST);
+ Schema schema = ops.loadTable("db1", "orc_variant").schema();
+ Types.NestedField payload = schema.findField("payload");
+ Assertions.assertEquals(Types.VariantType.get(), payload.type());
+ Assertions.assertEquals("new doc", payload.doc());
+ Assertions.assertTrue(payload.isOptional());
+ Assertions.assertEquals("payload", schema.columns().get(0).name());
+ }
+
+ @Test
+ public void testModifyColumnTypeInvolvingVariantFailsLoud() {
+ createTable("parquet_variant", "3", "parquet", true);
+ DorisConnectorException toString = Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.modifyColumn("db1", "parquet_variant",
+ change("payload", Types.StringType.get(), null, true), true, null));
+ Assertions.assertEquals("Cannot change column type involving Iceberg VARIANT: payload: variant -> string",
+ toString.getMessage());
+ DorisConnectorException toVariant = Assertions.assertThrows(DorisConnectorException.class,
+ () -> ops.modifyColumn("db1", "parquet_variant",
+ change("id", Types.VariantType.get(), null, true), true, null));
+ Assertions.assertEquals("Cannot change column type involving Iceberg VARIANT: id: long -> variant",
+ toVariant.getMessage());
+ Schema schema = ops.loadTable("db1", "parquet_variant").schema();
+ Assertions.assertEquals(Types.VariantType.get(), schema.findField("payload").type());
+ Assertions.assertEquals(Types.LongType.get(), schema.findField("id").type());
+ }
+
// ---------- reorderColumns ----------
@Test
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataDdlTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataDdlTest.java
index a767e72437f..b34147a12ce 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataDdlTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataDdlTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.connector.spi.ddl.PartitionFieldChange;
import org.apache.doris.connector.spi.ddl.TagChange;
import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -460,6 +461,52 @@ public class IcebergConnectorMetadataDdlTest {
Assertions.assertEquals(0, ctx.authCount);
}
+ @Test
+ public void testCreateTableWithVariantNeedsFormatThreeParquet() {
+ // A VARIANT column needs a format-version 3 table that writes Parquet data files, whether the table request
+ // or the catalog's table-default / table-override sets them; the check runs before the remote seam.
+ Map<String, String> v3Orc = new HashMap<>();
+ v3Orc.put(TableProperties.FORMAT_VERSION, "3");
+ v3Orc.put(TableProperties.DEFAULT_FILE_FORMAT, "orc");
+ assertVariantCreateRejected(Collections.emptyMap(), Collections.emptyMap(),
+ "Iceberg VARIANT writes require table format-version 3, but found 2");
+ assertVariantCreateRejected(v3Orc, Collections.emptyMap(),
+ "Iceberg VARIANT writes require Parquet data files, but found ORC");
+ assertVariantCreateRejected(Collections.singletonMap(TableProperties.FORMAT_VERSION, "3"),
+ Collections.singletonMap("table-override." + TableProperties.DEFAULT_FILE_FORMAT, "orc"),
+ "Iceberg VARIANT writes require Parquet data files, but found ORC");
+
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ Map<String, String> catalogProps = props(IcebergCatalogProperties.TYPE_REST);
+ catalogProps.put("table-default.format-version", "3");
+ new IcebergConnectorMetadata(ops, IcebergCatalogProperties.of(catalogProps), new RecordingConnectorContext())
+ .createTable(null, variantTable(Collections.emptyMap()));
+ Assertions.assertEquals(Types.VariantType.get(), ops.lastCreateSchema.findField("payload").type());
+ }
+
+ private static ConnectorCreateTableRequest variantTable(Map<String, String> properties) {
+ return ConnectorCreateTableRequest.builder()
+ .dbName("db1").tableName("t1")
+ .columns(Arrays.asList(
+ new ConnectorColumn("id", ConnectorType.of("BIGINT"), "", true, null, false),
+ new ConnectorColumn("payload", ConnectorType.of("VARIANT"), "", true, null, false)))
+ .properties(properties)
+ .build();
+ }
+
+ private static void assertVariantCreateRejected(Map<String, String> properties,
+ Map<String, String> catalogProperties, String message) {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ Map<String, String> catalogProps = props(IcebergCatalogProperties.TYPE_REST);
+ catalogProps.putAll(catalogProperties);
+ IcebergConnectorMetadata md = new IcebergConnectorMetadata(
+ ops, IcebergCatalogProperties.of(catalogProps), new RecordingConnectorContext());
+ DorisConnectorException ex = Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.createTable(null, variantTable(properties)));
+ Assertions.assertEquals(message, ex.getMessage());
+ Assertions.assertTrue(ops.log.isEmpty(), "reject must run before the remote seam");
+ }
+
// ---------- dropTable ----------
@Test
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilderTest.java
index 4eac0ff0abe..4c8a29b006a 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilderTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaBuilderTest.java
@@ -24,6 +24,8 @@ import org.apache.doris.connector.spi.ddl.ConnectorPartitionField;
import org.apache.doris.connector.spi.ddl.ConnectorPartitionSpec;
import org.apache.doris.connector.spi.ddl.ConnectorSortField;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.NullOrder;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -383,4 +385,70 @@ public class IcebergSchemaBuilderTest {
Map<String, String> props = IcebergSchemaBuilder.buildTableProperties(tableProps, catalogProps);
Assertions.assertEquals("1", props.get(TableProperties.FORMAT_VERSION));
}
+
+ @Test
+ public void variantIsSupportedOnlyAsTopLevelColumn() {
+ // A top-level VARIANT, as declared by DDL or as the VARIANT_COMPUTE_V2 type a CTAS reads from an Iceberg
+ // table, becomes an Iceberg VARIANT; one nested in a STRUCT, an ARRAY or a MAP is rejected, as is a
+ // nested field declared as VARIANT.
+ for (String typeName : Arrays.asList("VARIANT", "VARIANT_COMPUTE_V2")) {
+ ConnectorType variant = ConnectorType.of(typeName);
+ Schema schema = IcebergSchemaBuilder.buildSchema(Arrays.asList(
+ new ConnectorColumn("id", ConnectorType.of("INT"), null, false, null),
+ new ConnectorColumn("payload", variant, "doc", true, null)));
+ Assertions.assertEquals(Types.VariantType.get(), schema.findField("payload").type());
+ Assertions.assertTrue(schema.findField("payload").isOptional());
+ Assertions.assertEquals(Types.VariantType.get(), IcebergSchemaBuilder.buildColumnType(variant));
+ for (ConnectorType nested : Arrays.asList(
+ ConnectorType.structOf(Collections.singletonList("v"), Collections.singletonList(variant)),
+ ConnectorType.arrayOf(variant),
+ ConnectorType.mapOf(ConnectorType.of("STRING"), variant))) {
+ assertNestedVariantRejected(() -> IcebergSchemaBuilder.buildSchema(Collections.singletonList(
+ new ConnectorColumn("payload", nested, null, true, null))));
+ assertNestedVariantRejected(() -> IcebergSchemaBuilder.buildColumnType(nested));
+ }
+ assertNestedVariantRejected(() -> IcebergSchemaBuilder.buildNestedFieldType(variant));
+ }
+ }
+
+ private static void assertNestedVariantRejected(Runnable build) {
+ DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class, build::run);
+ Assertions.assertEquals("Iceberg VARIANT DDL currently supports only top-level columns",
+ exception.getMessage());
+ }
+
+ @Test
+ public void effectiveFileFormatFollowsCatalogAndTablePrecedence() {
+ // Catalog table-default, then the table request, then catalog table-override; the write-format nickname wins
+ // over write.format.default wherever it is set, as it does for an existing table.
+ Assertions.assertEquals(FileFormat.PARQUET,
+ IcebergSchemaBuilder.getEffectiveFileFormat(Collections.emptyMap(), Collections.emptyMap()));
+ Assertions.assertEquals(FileFormat.ORC, IcebergSchemaBuilder.getEffectiveFileFormat(
+ Collections.singletonMap(TableProperties.DEFAULT_FILE_FORMAT, "orc"), Collections.emptyMap()));
+ Assertions.assertEquals(FileFormat.ORC, IcebergSchemaBuilder.getEffectiveFileFormat(
+ Collections.emptyMap(), Collections.singletonMap(
+ CatalogProperties.TABLE_DEFAULT_PREFIX + TableProperties.DEFAULT_FILE_FORMAT, "orc")));
+ Assertions.assertEquals(FileFormat.PARQUET, IcebergSchemaBuilder.getEffectiveFileFormat(
+ Collections.singletonMap(TableProperties.DEFAULT_FILE_FORMAT, "orc"), Collections.singletonMap(
+ CatalogProperties.TABLE_OVERRIDE_PREFIX + TableProperties.DEFAULT_FILE_FORMAT, "parquet")));
+ Assertions.assertEquals(FileFormat.ORC, IcebergSchemaBuilder.getEffectiveFileFormat(
+ Collections.singletonMap("write-format", "orc"), Collections.singletonMap(
+ CatalogProperties.TABLE_OVERRIDE_PREFIX + TableProperties.DEFAULT_FILE_FORMAT, "parquet")));
+ Assertions.assertEquals(FileFormat.PARQUET, IcebergSchemaBuilder.getEffectiveFileFormat(
+ Collections.singletonMap(TableProperties.DEFAULT_FILE_FORMAT, "orc"), Collections.singletonMap(
+ CatalogProperties.TABLE_OVERRIDE_PREFIX + "write-format", "parquet")));
+ Map<String, String> catalogDefaults = new HashMap<>();
+ catalogDefaults.put(CatalogProperties.TABLE_DEFAULT_PREFIX + "write-format", "orc");
+ catalogDefaults.put(CatalogProperties.TABLE_DEFAULT_PREFIX + TableProperties.DEFAULT_FILE_FORMAT, "parquet");
+ Assertions.assertEquals(FileFormat.ORC,
+ IcebergSchemaBuilder.getEffectiveFileFormat(Collections.emptyMap(), catalogDefaults));
+ }
+
+ @Test
+ public void variantDefaultValuesMustBeNull() {
+ Assertions.assertNull(IcebergSchemaBuilder.parseDefaultLiteral(null, Types.VariantType.get()));
+ IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class,
+ () -> IcebergSchemaBuilder.parseDefaultLiteral("{\"source\":\"ddl\"}", Types.VariantType.get()));
+ Assertions.assertEquals("Iceberg VARIANT default values must be NULL", exception.getMessage());
+ }
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
index 0e7e3173678..323ac026b8a 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergSchemaUtilsTest.java
@@ -791,4 +791,14 @@ public class IcebergSchemaUtilsTest {
Assertions.assertThrows(RuntimeException.class,
() -> IcebergSchemaUtils.applySchemaEvolution(params, "!!!not-base64-thrift!!!"));
}
+
+ @Test
+ public void variantInitialDefaultMustBeNull() {
+ // A VARIANT added without an initial-default reads as NULL from older data files, and any other
+ // initial-default is rejected rather than handed to BE.
+ Assertions.assertNull(IcebergSchemaUtils.writeDefaultToDorisString(Types.VariantType.get(), null, false));
+ DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class,
+ () -> IcebergSchemaUtils.serializeInitialDefault(Types.VariantType.get(), "non-null-variant", false));
+ Assertions.assertEquals("Iceberg VARIANT initial-default must be NULL", exception.getMessage());
+ }
}
diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
index 621ef24ef79..8ab5566dc80 100644
--- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
+++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
@@ -113,47 +113,91 @@ public class IcebergWritePlanProviderTest {
Collections.singletonMap("iceberg.catalog.type", "hadoop");
@Test
- public void rejectsVariantDataWritesButAllowsDeleteOnlyMerge() {
- ConnectorColumn nestedVariant = new ConnectorColumn("payload",
- ConnectorType.structOf(Collections.singletonList("nested"),
- Collections.singletonList(ConnectorType.of("VARIANT"))),
- null, true, null);
- Assertions.assertThrows(DorisConnectorException.class,
- () -> IcebergWritePlanProvider.validateWriteSchema(
- Collections.singletonList(nestedVariant), true));
- Assertions.assertDoesNotThrow(() -> IcebergWritePlanProvider.validateWriteSchema(
- Collections.singletonList(nestedVariant), false));
-
- ConnectorColumn id = new ConnectorColumn(
- "id", ConnectorType.of("INT"), null, true, null);
- WriteHandle partialInsert = new WriteHandle(new IcebergTableHandle("db1", "t2"))
- .columns(Collections.singletonList(id))
- .boundTargetColumns(Arrays.asList(id, nestedVariant));
- Assertions.assertThrows(DorisConnectorException.class,
- () -> IcebergWritePlanProvider.validateWriteSchema(partialInsert));
+ public void variantDataWritesNeedUnshreddedFormatThreeParquetTables() {
+ // A write of data files whose target holds a VARIANT, at the root or nested, as plain VARIANT or as the
+ // VARIANT_COMPUTE_V2 carrier the engine hands to the write path, needs format-version 3 and unshredded
+ // Parquet data files. A write without data files (DELETE, a delete-only MERGE) and a target without a
+ // VARIANT are not checked. MUTATION: dropping the VARIANT_COMPUTE_V2 arm of containsVariant -> red.
+ ConnectorType carrier = ConnectorType.of("VARIANT_COMPUTE_V2");
+ ConnectorColumn id = new ConnectorColumn("id", ConnectorType.of("INT"), null, true, null);
+ List<ConnectorColumn> variants = Arrays.asList(
+ new ConnectorColumn("v", ConnectorType.of("VARIANT"), null, true, null),
+ new ConnectorColumn("v", carrier, null, true, null),
+ new ConnectorColumn("payload", ConnectorType.structOf(Collections.singletonList("nested"),
+ Collections.singletonList(carrier)), null, true, null));
+ Map<String, String> unshredded = Collections.emptyMap();
+ for (ConnectorColumn variant : variants) {
+ List<ConnectorColumn> columns = Arrays.asList(id, variant);
+ Assertions.assertDoesNotThrow(() -> IcebergWritePlanProvider.validateVariantWrite(
+ columns, true, 3, FileFormat.PARQUET, unshredded), variant.getType().toString());
+ assertVariantWriteRejected(columns, 2, FileFormat.PARQUET, unshredded,
+ "Iceberg VARIANT writes require table format-version 3, but found 2");
+ assertVariantWriteRejected(columns, 3, FileFormat.ORC, unshredded,
+ "Iceberg VARIANT writes require Parquet data files, but found ORC");
+ assertVariantWriteRejected(columns, 3, FileFormat.PARQUET,
+ Collections.singletonMap("write.parquet.shred-variants", "true"),
+ "Doris currently supports only unshredded Iceberg VARIANT writes");
+ Assertions.assertDoesNotThrow(() -> IcebergWritePlanProvider.validateVariantWrite(
+ columns, false, 2, FileFormat.ORC, unshredded), variant.getType().toString());
+ }
+ Assertions.assertDoesNotThrow(() -> IcebergWritePlanProvider.validateVariantWrite(
+ Collections.singletonList(id), true, 2, FileFormat.ORC, unshredded));
+ }
+
+ private static void assertVariantWriteRejected(List<ConnectorColumn> columns, int formatVersion,
+ FileFormat fileFormat, Map<String, String> writerProperties, String message) {
+ DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class,
+ () -> IcebergWritePlanProvider.validateVariantWrite(
+ columns, true, formatVersion, fileFormat, writerProperties));
+ Assertions.assertTrue(exception.getMessage().contains(message), exception.getMessage());
}
@Test
- public void rejectsRootAndNestedComputeVariantCarrier() {
- // The engine hands an Iceberg VARIANT column to the write path as the VARIANT_COMPUTE_V2 carrier
- // (ConnectorColumnConverter), not as plain VARIANT, so the read-only gate must recognize the carrier
- // both as a root column and nested in a complex type. MUTATION: dropping the VARIANT_COMPUTE_V2 arm
- // of containsVariant -> red.
- ConnectorType carrier = ConnectorType.of("VARIANT_COMPUTE_V2");
- ConnectorColumn rootVariant = new ConnectorColumn("v", carrier, null, true, null);
- ConnectorColumn nestedVariant = new ConnectorColumn("payload",
- ConnectorType.structOf(Collections.singletonList("nested"), Collections.singletonList(carrier)),
- null, true, null);
- for (ConnectorColumn column : Arrays.asList(rootVariant, nestedVariant)) {
- DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class,
- () -> IcebergWritePlanProvider.validateWriteSchema(Collections.singletonList(column), true),
- column.getName());
- Assertions.assertTrue(exception.getMessage().contains("VARIANT")
- && exception.getMessage().contains("read-only"), exception.getMessage());
- // A delete-only MERGE writes no data file, so the same schema stays usable for it.
- Assertions.assertDoesNotThrow(() -> IcebergWritePlanProvider.validateWriteSchema(
- Collections.singletonList(column), false), column.getName());
- }
+ public void planWriteWritesVariantsOnlyToFormatThreeParquetTables() {
+ // An INSERT whose column list omits the VARIANT column still binds the whole target schema, so the table it
+ // writes is checked. A v3 Parquet table takes the write, with the VARIANT in the schema JSON BE parses and
+ // column metrics collected for it although it is the only column with metrics, while a v3 ORC table and a
+ // shredding Parquet table reject it. MUTATION: collecting metrics only for primitive fields -> red.
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "payload", Types.VariantType.get()));
+ InMemoryCatalog catalog = freshCatalog();
+ Map<String, String> props = new HashMap<>();
+ props.put("format-version", "3");
+ props.put("write.format.default", "parquet");
+ props.put("write.data.path", "oss://bucket/wh/db1/variants/data");
+ props.put("write.metadata.metrics.default", "none");
+ props.put("write.metadata.metrics.column.payload", "counts");
+ Table parquet = catalog.createTable(TableIdentifier.of("db1", "variants"), schema,
+ PartitionSpec.unpartitioned(), props);
+ TIcebergTableSink sink = planSink(parquet, contextWithStorage(), variantInsert(parquet, "variants"));
+ Assertions.assertTrue(sink.getSchemaJson().contains("\"type\":\"variant\""), sink.getSchemaJson());
+ Assertions.assertTrue(sink.isCollectColumnStats());
+
+ props.put("write.format.default", "orc");
+ Table orc = catalog.createTable(TableIdentifier.of("db1", "orc_variants"), schema,
+ PartitionSpec.unpartitioned(), props);
+ DorisConnectorException orcError = Assertions.assertThrows(DorisConnectorException.class,
+ () -> planSink(orc, contextWithStorage(), variantInsert(orc, "orc_variants")));
+ Assertions.assertTrue(orcError.getMessage().contains("require Parquet data files, but found ORC"),
+ orcError.getMessage());
+
+ props.put("write.format.default", "parquet");
+ props.put("write.parquet.shred-variants", "true");
+ Table shredded = catalog.createTable(TableIdentifier.of("db1", "shredded_variants"), schema,
+ PartitionSpec.unpartitioned(), props);
+ DorisConnectorException shredError = Assertions.assertThrows(DorisConnectorException.class,
+ () -> planSink(shredded, contextWithStorage(), variantInsert(shredded, "shredded_variants")));
+ Assertions.assertTrue(shredError.getMessage().contains("unshredded"), shredError.getMessage());
+ }
+
+ /** An INSERT into {@code id} only, bound to the whole write schema of the table. */
+ private static WriteHandle variantInsert(Table table, String tableName) {
+ RecordingConnectorContext context = contextWithStorage();
+ IcebergTableHandle tableHandle = new IcebergTableHandle("db1", tableName);
+ List<ConnectorColumn> columns = providerFor(table, context).getWriteColumns(
+ sessionFor(table, context), tableHandle, Optional.empty()).orElseThrow(AssertionError::new);
+ return new WriteHandle(tableHandle).columns(columns.subList(0, 1)).boundTargetColumns(columns);
}
private static InMemoryCatalog freshCatalog() {
@@ -381,6 +425,33 @@ public class IcebergWritePlanProviderTest {
IcebergWriteSchemaContext.toDorisSql(structType, structDefault, false, false));
}
+ @Test
+ public void variantWriteDefaultsMustBeNull() {
+ // An optional VARIANT column without a write-default writes NULL and a required one has no default, while a
+ // VARIANT write-default other than NULL is rejected; a STRUCT default keeps its VARIANT child NULL.
+ Schema schema = new Schema(
+ Types.NestedField.optional(1, "optional_payload", Types.VariantType.get()),
+ Types.NestedField.required(2, "required_payload", Types.VariantType.get()));
+ Table table = freshCatalog().createTable(TableIdentifier.of("db1", "variant_defaults"), schema,
+ PartitionSpec.unpartitioned(), Collections.singletonMap("format-version", "3"));
+ IcebergWriteSchemaContext writeContext = IcebergWriteSchemaContext.create(
+ table, "db1.variant_defaults", Optional.empty(), false, false);
+ Assertions.assertEquals("NULL", writeContext.getColumns().get(0).getDefaultValueSql());
+ Assertions.assertNull(writeContext.getColumns().get(1).getDefaultValueSql());
+
+ DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class,
+ () -> IcebergWriteSchemaContext.toDorisSql(Types.VariantType.get(), "{}", false, false));
+ Assertions.assertTrue(exception.getMessage().contains("Iceberg VARIANT write-default must be NULL"),
+ exception.getMessage());
+
+ Types.StructType structType = Types.StructType.of(
+ Types.NestedField.optional(3, "payload", Types.VariantType.get()));
+ GenericRecord structDefault = GenericRecord.create(structType);
+ structDefault.set(0, null);
+ Assertions.assertEquals("named_struct('payload', NULL)",
+ IcebergWriteSchemaContext.toDorisSql(structType, structDefault, false, false));
+ }
+
@Test
public void branchWritePinsBranchHeadSchema() {
InMemoryCatalog catalog = freshCatalog();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java
new file mode 100644
index 00000000000..36a02ca14f1
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/VariantWritePlanValidator.java
@@ -0,0 +1,529 @@
+// 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.
+
+package org.apache.doris.datasource;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.CTEContext;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.CTEId;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.ExprId;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.OrderExpression;
+import org.apache.doris.nereids.trees.expressions.ScalarSubquery;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.WhenClause;
+import org.apache.doris.nereids.trees.expressions.WindowExpression;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.algebra.RecursiveCte;
+import org.apache.doris.nereids.trees.plans.algebra.SetOperation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalExcept;
+import org.apache.doris.nereids.trees.plans.logical.LogicalGenerate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalIntersect;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRecursiveUnionAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWorkTableReference;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.MapType;
+import org.apache.doris.nereids.types.StructField;
+import org.apache.doris.nereids.types.StructType;
+import org.apache.doris.nereids.types.VariantType;
+
+import com.google.common.collect.ImmutableSet;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/** Shared source-plan validation for external Variant sinks. */
+public final class VariantWritePlanValidator {
+ private VariantWritePlanValidator() {
+ }
+
+ /**
+ * Rejects an implicit Variant-to-non-Variant cast in the lineage of a Variant target column.
+ *
+ * <p>Common-type analysis for UNION, IF and CASE runs before sink binding. Without this
+ * check, an object or array Variant can become SQL NULL while being cast to a scalar, and the
+ * sink will only see that scalar/NULL and encode it back as Variant. Explicit casts remain an
+ * intentional user conversion and are not rejected. Only the expressions whose values can reach
+ * the target are inspected: the condition of an IF or of a CASE branch only chooses a value, so a
+ * Variant path compared with a scalar there loses nothing that is written. Likewise the later
+ * inputs of EXCEPT and INTERSECT only decide which rows of the first input stay, and the partition
+ * and order keys of a window and the ORDER BY of an aggregate function only arrange the rows whose
+ * values are used.</p>
+ */
+ public static void validateNoLossyCoercion(List<Column> targetColumns, Plan sourcePlan) {
+ validateNoLossyCoercion(targetColumns, sourcePlan, Optional.empty());
+ }
+
+ /** Validates a sink while its enclosing CTE producers still live in the analyzer context. */
+ public static void validateNoLossyCoercion(
+ List<Column> targetColumns, Plan sourcePlan, CTEContext cteContext) {
+ validateNoLossyCoercion(targetColumns, sourcePlan, Optional.of(cteContext));
+ }
+
+ private static void validateNoLossyCoercion(
+ List<Column> targetColumns, Plan sourcePlan, Optional<CTEContext> cteContext) {
+ if (targetColumns.size() != sourcePlan.getOutput().size()) {
+ throw new AnalysisException("VARIANT write target and source columns are not aligned");
+ }
+ List<Integer> sourceOrdinals = new ArrayList<>(targetColumns.size());
+ for (int i = 0; i < targetColumns.size(); i++) {
+ sourceOrdinals.add(i);
+ }
+ validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext);
+ }
+
+ /**
+ * Validates target columns that are embedded in a wider source output, such as row-level DML
+ * where the operation and row-id routing columns precede the table data columns.
+ */
+ public static void validateNoLossyCoercion(
+ List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals) {
+ validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.empty());
+ }
+
+ /** Validates mapped source ordinals with analyzer-time CTE producer visibility. */
+ public static void validateNoLossyCoercion(
+ List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals,
+ CTEContext cteContext) {
+ validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, Optional.of(cteContext));
+ }
+
+ private static void validateNoLossyCoercion(
+ List<Column> targetColumns, Plan sourcePlan, List<Integer> sourceOrdinals,
+ Optional<CTEContext> cteContext) {
+ if (targetColumns.size() != sourceOrdinals.size()) {
+ throw new AnalysisException("VARIANT write target and source columns are not aligned");
+ }
+
+ List<Integer> variantTargetOrdinals = new ArrayList<>();
+ for (int i = 0; i < targetColumns.size(); i++) {
+ DataType targetType = DataType.fromCatalogType(targetColumns.get(i).getType());
+ if (VariantType.containsVariant(targetType)) {
+ variantTargetOrdinals.add(i);
+ }
+ }
+ if (variantTargetOrdinals.isEmpty()) {
+ return;
+ }
+
+ TraceContext traceContext = new TraceContext(sourcePlan, cteContext);
+ List<Slot> sourceOutputs = sourcePlan.getOutput();
+ for (int targetOrdinal : variantTargetOrdinals) {
+ int sourceOrdinal = sourceOrdinals.get(targetOrdinal);
+ if (sourceOrdinal < 0 || sourceOrdinal >= sourceOutputs.size()) {
+ throw new AnalysisException("VARIANT write target and source columns are not aligned");
+ }
+ Column targetColumn = targetColumns.get(targetOrdinal);
+ traceOutputLineage(
+ sourcePlan,
+ Collections.singleton(sourceOutputs.get(sourceOrdinal).getExprId()),
+ targetColumn.getName(),
+ traceContext);
+ }
+ }
+
+ /**
+ * Validates the requested outputs and returns lineage ExprIds supplied by an outer scope.
+ * Scalar subqueries use the returned IDs to continue tracing correlated columns in the
+ * enclosing plan.
+ */
+ private static Set<ExprId> traceOutputLineage(
+ Plan plan, Set<ExprId> requiredExprIds, String targetColumn, TraceContext context) {
+ if (requiredExprIds.isEmpty()) {
+ return Collections.emptySet();
+ }
+ if (plan instanceof SetOperation) {
+ return traceSetOperation(plan, (SetOperation) plan, requiredExprIds, targetColumn, context);
+ }
+ if (plan instanceof RecursiveCte) {
+ return traceRecursiveCte(plan, (RecursiveCte) plan, requiredExprIds, targetColumn, context);
+ }
+ if (plan instanceof LogicalGenerate) {
+ return traceGenerate((LogicalGenerate<?>) plan, requiredExprIds, targetColumn, context);
+ }
+ if (plan instanceof LogicalCTEConsumer) {
+ return traceCteConsumer((LogicalCTEConsumer) plan, requiredExprIds, targetColumn, context);
+ }
+
+ Set<ExprId> unresolvedExprIds = new HashSet<>(requiredExprIds);
+ Set<ExprId> inputExprIds = new HashSet<>();
+ for (Expression expression : plan.getExpressions()) {
+ if (!(expression instanceof NamedExpression)) {
+ continue;
+ }
+ NamedExpression namedExpression = (NamedExpression) expression;
+ if (!requiredExprIds.contains(namedExpression.getExprId())) {
+ continue;
+ }
+ inputExprIds.addAll(validateExpression(namedExpression, targetColumn, context));
+ unresolvedExprIds.remove(namedExpression.getExprId());
+ }
+ inputExprIds.addAll(unresolvedExprIds);
+
+ Set<ExprId> externalExprIds = new HashSet<>(inputExprIds);
+ for (Plan child : plan.children()) {
+ Set<ExprId> childOutputExprIds = child.getOutputExprIdSet();
+ Set<ExprId> childRequiredExprIds = new HashSet<>(inputExprIds);
+ childRequiredExprIds.retainAll(childOutputExprIds);
+ externalExprIds.removeAll(childOutputExprIds);
+ externalExprIds.addAll(traceOutputLineage(child, childRequiredExprIds, targetColumn, context));
+ }
+ return externalExprIds;
+ }
+
+ private static Set<ExprId> traceRecursiveCte(
+ Plan plan, RecursiveCte recursiveCte, Set<ExprId> requiredExprIds,
+ String targetColumn, TraceContext context) {
+ // The recursive term reads the previous iteration through the work table, so a column it writes
+ // can carry the value of another column. Trace those columns as well until no new one is reached.
+ Map<ExprId, Integer> workTableOrdinals = workTableOrdinals(plan);
+ Set<ExprId> externalExprIds = new HashSet<>();
+ Set<Integer> tracedOrdinals = new HashSet<>();
+ Set<Integer> pendingOrdinals = requiredOutputOrdinals(plan, requiredExprIds);
+ while (!pendingOrdinals.isEmpty()) {
+ tracedOrdinals.addAll(pendingOrdinals);
+ Set<ExprId> childExternalExprIds = new HashSet<>();
+ for (int childIndex = 0; childIndex < plan.arity(); childIndex++) {
+ List<SlotReference> childOutputs = recursiveCte.getRegularChildOutput(childIndex);
+ Set<ExprId> childRequiredExprIds = new HashSet<>();
+ for (int ordinal : pendingOrdinals) {
+ if (ordinal < childOutputs.size()) {
+ childRequiredExprIds.add(childOutputs.get(ordinal).getExprId());
+ }
+ }
+ childExternalExprIds.addAll(traceOutputLineage(
+ plan.child(childIndex), childRequiredExprIds, targetColumn, context));
+ }
+ pendingOrdinals = new HashSet<>();
+ for (ExprId exprId : childExternalExprIds) {
+ Integer ordinal = workTableOrdinals.get(exprId);
+ if (ordinal == null) {
+ externalExprIds.add(exprId);
+ } else if (!tracedOrdinals.contains(ordinal)) {
+ pendingOrdinals.add(ordinal);
+ }
+ }
+ }
+ return externalExprIds;
+ }
+
+ /** Maps each column of the work table the recursive term reads to the CTE output ordinal it holds. */
+ private static Map<ExprId, Integer> workTableOrdinals(Plan recursiveCte) {
+ CTEId cteId = ((LogicalRecursiveUnionAnchor<?>) recursiveCte.child(0)).getCteId();
+ List<LogicalWorkTableReference> workTables = recursiveCte.child(1).collectToList(
+ node -> node instanceof LogicalWorkTableReference
+ && ((LogicalWorkTableReference) node).getCteId().equals(cteId));
+ Map<ExprId, Integer> ordinals = new HashMap<>();
+ for (LogicalWorkTableReference workTable : workTables) {
+ List<Slot> outputs = workTable.getOutput();
+ for (int ordinal = 0; ordinal < outputs.size(); ordinal++) {
+ ordinals.put(outputs.get(ordinal).getExprId(), ordinal);
+ }
+ }
+ return ordinals;
+ }
+
+ private static Set<ExprId> traceGenerate(
+ LogicalGenerate<?> generate, Set<ExprId> requiredExprIds,
+ String targetColumn, TraceContext context) {
+ Set<ExprId> childRequiredExprIds = new HashSet<>(requiredExprIds);
+ List<Slot> generatorOutputs = generate.getGeneratorOutput();
+ List<? extends Expression> generators = generate.getGenerators();
+ for (int ordinal = 0; ordinal < generatorOutputs.size(); ordinal++) {
+ Slot generatorOutput = generatorOutputs.get(ordinal);
+ if (!requiredExprIds.contains(generatorOutput.getExprId())) {
+ continue;
+ }
+ childRequiredExprIds.remove(generatorOutput.getExprId());
+ childRequiredExprIds.addAll(validateExpression(generators.get(ordinal), targetColumn, context));
+ }
+ Set<ExprId> childOutputExprIds = generate.child().getOutputExprIdSet();
+ Set<ExprId> externalExprIds = new HashSet<>(childRequiredExprIds);
+ externalExprIds.removeAll(childOutputExprIds);
+ childRequiredExprIds.retainAll(childOutputExprIds);
+ externalExprIds.addAll(traceOutputLineage(
+ generate.child(), childRequiredExprIds, targetColumn, context));
+ return externalExprIds;
+ }
+
+ private static Set<ExprId> traceSetOperation(
+ Plan plan, SetOperation setOperation, Set<ExprId> requiredExprIds,
+ String targetColumn, TraceContext context) {
+ Set<ExprId> externalExprIds = new HashSet<>();
+ Set<Integer> requiredOrdinals = requiredOutputOrdinals(plan, requiredExprIds);
+
+ if (plan instanceof LogicalUnion) {
+ for (List<NamedExpression> constantRow : ((LogicalUnion) plan).getConstantExprsList()) {
+ for (int ordinal : requiredOrdinals) {
+ if (ordinal < constantRow.size()) {
+ externalExprIds.addAll(validateExpression(constantRow.get(ordinal), targetColumn, context));
+ }
+ }
+ }
+ }
+
+ // EXCEPT and INTERSECT return rows of their first input; the later inputs only decide which of
+ // those rows stay, so none of their values is written.
+ int valueInputs = plan instanceof LogicalExcept || plan instanceof LogicalIntersect
+ ? 1
+ : setOperation.getArity();
+ for (int childIndex = 0; childIndex < valueInputs; childIndex++) {
+ List<SlotReference> childOutputs = setOperation.getRegularChildOutput(childIndex);
+ Set<ExprId> childRequiredExprIds = new HashSet<>();
+ for (int ordinal : requiredOrdinals) {
+ if (ordinal < childOutputs.size()) {
+ childRequiredExprIds.add(childOutputs.get(ordinal).getExprId());
+ }
+ }
+ externalExprIds.addAll(traceOutputLineage(
+ plan.child(childIndex), childRequiredExprIds, targetColumn, context));
+ }
+ return externalExprIds;
+ }
+
+ private static Set<Integer> requiredOutputOrdinals(Plan plan, Set<ExprId> requiredExprIds) {
+ Set<Integer> requiredOrdinals = new HashSet<>();
+ List<Slot> outputs = plan.getOutput();
+ for (int i = 0; i < outputs.size(); i++) {
+ if (requiredExprIds.contains(outputs.get(i).getExprId())) {
+ requiredOrdinals.add(i);
+ }
+ }
+ return requiredOrdinals;
+ }
+
+ private static Set<ExprId> traceCteConsumer(
+ LogicalCTEConsumer consumer, Set<ExprId> requiredExprIds,
+ String targetColumn, TraceContext context) {
+ // A producer shared by several consumers is traced once per output, whichever sets of its outputs
+ // the consumers request, so the work stays proportional to the plan rather than to the number of
+ // paths through it or of the sets of outputs requested along them.
+ Set<ExprId> externalExprIds = new HashSet<>();
+ for (Map.Entry<Slot, Slot> mapping : consumer.getConsumerToProducerOutputMap().entrySet()) {
+ if (requiredExprIds.contains(mapping.getKey().getExprId())) {
+ externalExprIds.addAll(traceCteOutput(
+ consumer, mapping.getValue().getExprId(), targetColumn, context));
+ }
+ }
+ return externalExprIds;
+ }
+
+ private static Set<ExprId> traceCteOutput(
+ LogicalCTEConsumer consumer, ExprId producerExprId, String targetColumn, TraceContext context) {
+ Pair<CTEId, ExprId> traceKey = Pair.of(consumer.getCteId(), producerExprId);
+ Set<ExprId> completedTrace = context.completedCteTraces.get(traceKey);
+ if (completedTrace != null) {
+ return completedTrace;
+ }
+ LogicalCTEProducer<?> producer = context.cteProducers.get(consumer.getCteId());
+ Optional<Plan> producerPlan;
+ if (producer == null) {
+ // A producer taken from the analyzer context is not part of the traced plan, so the
+ // producers of the WITH clauses nested in it have not been collected yet.
+ producerPlan = context.findAnalyzerCteProducer(consumer);
+ producerPlan.ifPresent(context::collectCteProducers);
+ } else {
+ producerPlan = Optional.of(producer.child());
+ }
+ if (!producerPlan.isPresent() || !context.activeCteIds.add(consumer.getCteId())) {
+ return Collections.emptySet();
+ }
+ try {
+ Set<ExprId> externalExprIds = ImmutableSet.copyOf(traceOutputLineage(
+ producerPlan.get(), Collections.singleton(producerExprId), targetColumn, context));
+ context.completedCteTraces.put(traceKey, externalExprIds);
+ return externalExprIds;
+ } finally {
+ context.activeCteIds.remove(consumer.getCteId());
+ }
+ }
+
+ /**
+ * Validates the casts on the values {@code expression} writes and returns the ExprIds those
+ * values come from: the slots they read, and the correlated slots of their scalar subqueries,
+ * which an outer scope may supply.
+ */
+ private static Set<ExprId> validateExpression(
+ Expression expression, String targetColumn, TraceContext context) {
+ List<Expression> valueNodes = valueNodes(expression);
+ validateLossyCast(valueNodes, targetColumn);
+ Set<ExprId> inputExprIds = new HashSet<>();
+ for (Expression node : valueNodes) {
+ if (node instanceof Slot) {
+ inputExprIds.add(((Slot) node).getExprId());
+ } else if (node instanceof ScalarSubquery) {
+ inputExprIds.addAll(traceScalarSubquery((ScalarSubquery) node, targetColumn, context));
+ }
+ }
+ return inputExprIds;
+ }
+
+ /** Traces a scalar subquery and returns its correlated slots that the subquery itself does not supply. */
+ private static Set<ExprId> traceScalarSubquery(
+ ScalarSubquery scalarSubquery, String targetColumn, TraceContext context) {
+ scalarSubquery.getTypeCoercionExpr()
+ .ifPresent(coercion -> validateLossyCast(valueNodes(coercion), targetColumn));
+ Plan queryPlan = scalarSubquery.getQueryPlan();
+ context.collectCteProducers(queryPlan);
+ Set<ExprId> unresolvedExprIds = new HashSet<>(traceOutputLineage(
+ queryPlan,
+ Collections.singleton(queryPlan.getOutput().get(0).getExprId()),
+ targetColumn,
+ context));
+ Set<ExprId> correlatedExprIds = new HashSet<>();
+ for (Slot correlateSlot : scalarSubquery.getCorrelateSlots()) {
+ correlatedExprIds.add(correlateSlot.getExprId());
+ }
+ unresolvedExprIds.retainAll(correlatedExprIds);
+ return unresolvedExprIds;
+ }
+
+ /**
+ * The nodes of {@code expression} whose values can reach its result, in pre-order: every node
+ * except those under the condition of an IF or of a CASE branch, which only chooses a value, and
+ * those under the partition and order keys of a window or the ORDER BY of an aggregate function
+ * such as GROUP_CONCAT, which only arrange the rows whose values are used.
+ */
+ private static List<Expression> valueNodes(Expression expression) {
+ List<Expression> nodes = new ArrayList<>();
+ collectValueNodes(expression, nodes);
+ return nodes;
+ }
+
+ private static void collectValueNodes(Expression expression, List<Expression> nodes) {
+ nodes.add(expression);
+ if (expression instanceof If) {
+ collectValueNodes(((If) expression).getTrueValue(), nodes);
+ collectValueNodes(((If) expression).getFalseValue(), nodes);
+ } else if (expression instanceof WhenClause) {
+ collectValueNodes(((WhenClause) expression).getResult(), nodes);
+ } else if (expression instanceof WindowExpression) {
+ collectValueNodes(((WindowExpression) expression).getFunction(), nodes);
+ } else {
+ for (Expression child : expression.children()) {
+ if (!(child instanceof OrderExpression)) {
+ collectValueNodes(child, nodes);
+ }
+ }
+ }
+ }
+
+ private static void validateLossyCast(List<Expression> valueNodes, String targetColumn) {
+ for (Expression node : valueNodes) {
+ if (!(node instanceof Cast)) {
+ continue;
+ }
+ Cast cast = (Cast) node;
+ if (!cast.isExplicitType() && losesVariantLeaf(cast.child().getDataType(), cast.getDataType())) {
+ throw new AnalysisException(
+ "VARIANT write cannot safely convert input column '" + targetColumn
+ + "': source expression implicitly casts VARIANT to "
+ + cast.getDataType().toSql() + " before sink analysis");
+ }
+ }
+ }
+
+ private static boolean losesVariantLeaf(DataType sourceType, DataType targetType) {
+ // Encoding a scalar or supported ARRAY value as one top-level Variant preserves all
+ // nested Variant leaves. The lossy case is a Variant leaf coerced to a non-Variant path.
+ if (targetType instanceof VariantType) {
+ return false;
+ }
+ if (sourceType instanceof VariantType) {
+ return true;
+ }
+ if (sourceType instanceof ArrayType) {
+ return targetType instanceof ArrayType
+ ? losesVariantLeaf(
+ ((ArrayType) sourceType).getItemType(),
+ ((ArrayType) targetType).getItemType())
+ : VariantType.containsVariant(sourceType);
+ }
+ if (sourceType instanceof MapType) {
+ if (!(targetType instanceof MapType)) {
+ return VariantType.containsVariant(sourceType);
+ }
+ MapType sourceMap = (MapType) sourceType;
+ MapType targetMap = (MapType) targetType;
+ return losesVariantLeaf(sourceMap.getKeyType(), targetMap.getKeyType())
+ || losesVariantLeaf(sourceMap.getValueType(), targetMap.getValueType());
+ }
+ if (sourceType instanceof StructType) {
+ if (!(targetType instanceof StructType)) {
+ return VariantType.containsVariant(sourceType);
+ }
+ List<StructField> sourceFields = ((StructType) sourceType).getFields();
+ List<StructField> targetFields = ((StructType) targetType).getFields();
+ if (sourceFields.size() != targetFields.size()) {
+ return VariantType.containsVariant(sourceType);
+ }
+ for (int i = 0; i < sourceFields.size(); i++) {
+ if (losesVariantLeaf(sourceFields.get(i).getDataType(), targetFields.get(i).getDataType())) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ private static final class TraceContext {
+ private final Map<CTEId, LogicalCTEProducer<?>> cteProducers = new HashMap<>();
+ private final Set<CTEId> activeCteIds = new HashSet<>();
+ // Completed traces of producer outputs, keyed by CTE and output, with the ExprIds they return.
+ private final Map<Pair<CTEId, ExprId>, Set<ExprId>> completedCteTraces = new HashMap<>();
+ private final Optional<CTEContext> analyzerCteContext;
+
+ private TraceContext(Plan sourcePlan, Optional<CTEContext> analyzerCteContext) {
+ this.analyzerCteContext = analyzerCteContext;
+ collectCteProducers(sourcePlan);
+ }
+
+ private Optional<Plan> findAnalyzerCteProducer(LogicalCTEConsumer consumer) {
+ return analyzerCteContext
+ .flatMap(context -> context.findCTEContext(consumer.getName()))
+ .filter(context -> context.getCteId().equals(consumer.getCteId()))
+ .flatMap(context -> context.getAnalyzedCTEPlan(consumer.getName()))
+ .map(Plan.class::cast);
+ }
+
+ private void collectCteProducers(Plan plan) {
+ if (plan instanceof LogicalCTEProducer) {
+ LogicalCTEProducer<?> producer = (LogicalCTEProducer<?>) plan;
+ cteProducers.put(producer.getCteId(), producer);
+ }
+ for (Plan child : plan.children()) {
+ collectCteProducers(child);
+ }
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java
index 49833b0b5e1..456b7ca1892 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindExpression.java
@@ -21,6 +21,8 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.FunctionRegistry;
import org.apache.doris.common.Pair;
+import org.apache.doris.datasource.VariantWritePlanValidator;
+import org.apache.doris.nereids.CTEContext;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.SqlCacheContext;
import org.apache.doris.nereids.StatementContext;
@@ -298,76 +300,80 @@ public class BindExpression implements AnalysisRuleFactory {
List<NamedExpression> outputExprs = sink.child().getOutput().stream()
.map(NamedExpression.class::cast)
.collect(ImmutableList.toImmutableList());
- List<Column> visibleColumns = sink.getCols().stream()
- .filter(Column::isVisible)
- .collect(ImmutableList.toImmutableList());
- // The connector-reserved passthrough columns (iceberg v3 row-lineage) are the hidden target columns
- // marked reservedPassthrough. Derive their names from the sink's target schema so the meta-column
- // check below stays name-based here (this site sees output-expression names, not Column objects) while
- // fe-core no longer string-matches the iceberg column names.
- List<String> reservedPassthroughNames = sink.getCols().stream()
- .filter(Column::isReservedPassthrough)
- .map(Column::getName)
- .collect(ImmutableList.toImmutableList());
- int dataExprCount = 0;
- for (NamedExpression expr : outputExprs) {
- if (!isIcebergMergeMetaColumn(expr.getName(), reservedPassthroughNames)) {
- dataExprCount++;
- }
+ if (sink.isWritesDataFiles()) {
+ validateMergeNoLossyCoercion(sink.getCols(), sink.child(), ctx.cascadesContext.getCteContext());
}
- if (dataExprCount != visibleColumns.size()) {
+ List<NamedExpression> castExprs = coerceMergeOutput(sink.getCols(), outputExprs);
+ if (castExprs.equals(outputExprs)) {
if (sink.getOutputExprs().equals(outputExprs)) {
return sink;
}
return sink.withOutputExprs(outputExprs);
}
+ LogicalProject<?> project = new LogicalProject<>(castExprs, sink.child());
+ return (LogicalExternalRowLevelMergeSink<Plan>) sink.withChildAndUpdateOutput(project);
+ }
- boolean changed = false;
- int columnIndex = 0;
- List<NamedExpression> castExprs = Lists.newArrayListWithCapacity(outputExprs.size());
- for (NamedExpression expr : outputExprs) {
- if (isIcebergMergeMetaColumn(expr.getName(), reservedPassthroughNames)) {
- castExprs.add(expr);
+ /**
+ * Rejects a lossy implicit cast in the lineage of a Variant target column of a merge sink that
+ * writes data files. See {@link #coerceMergeOutput} for the layout of the sink output.
+ */
+ static void validateMergeNoLossyCoercion(
+ List<Column> sinkColumns, Plan sourcePlan, CTEContext cteContext) {
+ List<Column> targetColumns = Lists.newArrayList();
+ List<Integer> sourceOrdinals = Lists.newArrayList();
+ int sourceOrdinal = 2;
+ for (Column column : sinkColumns) {
+ if (!column.isVisible() && !column.isReservedPassthrough()) {
continue;
}
- Column column = visibleColumns.get(columnIndex++);
- DataType targetType = DataType.fromCatalogType(column.getType());
- Expression castExpr = TypeCoercionUtils.castIfNotSameType(expr, targetType);
- NamedExpression namedExpr;
- if (castExpr instanceof NamedExpression) {
- namedExpr = (NamedExpression) castExpr;
- if (!column.getName().equalsIgnoreCase(namedExpr.getName())) {
- namedExpr = new Alias(namedExpr, column.getName());
- }
- } else {
- namedExpr = new Alias(castExpr, column.getName());
- }
- if (!namedExpr.equals(expr)) {
- changed = true;
- }
- castExprs.add(namedExpr);
- }
- if (!changed) {
- if (sink.getOutputExprs().equals(outputExprs)) {
- return sink;
+ if (column.isVisible()) {
+ targetColumns.add(column);
+ sourceOrdinals.add(sourceOrdinal);
}
- return sink.withOutputExprs(outputExprs);
+ sourceOrdinal++;
}
- LogicalProject<?> project = new LogicalProject<>(castExprs, sink.child());
- return (LogicalExternalRowLevelMergeSink<Plan>) sink.withChildAndUpdateOutput(project);
+ VariantWritePlanValidator.validateNoLossyCoercion(targetColumns, sourcePlan, sourceOrdinals, cteContext);
}
- private boolean isIcebergMergeMetaColumn(String name, List<String> reservedPassthroughNames) {
- if (MergeOperation.OPERATION_COLUMN.equalsIgnoreCase(name)) {
- return true;
+ /**
+ * Casts the visible data columns of a merge sink output to their target types. The output is the
+ * operation and row-id routing columns followed by one column per visible or connector-reserved
+ * passthrough target column, in target order. Columns are matched by position rather than by name,
+ * since a data column may be named like a routing column, such as `operation`.
+ */
+ static List<NamedExpression> coerceMergeOutput(List<Column> sinkColumns, List<NamedExpression> outputExprs) {
+ List<Column> outputColumns = sinkColumns.stream()
+ .filter(column -> column.isVisible() || column.isReservedPassthrough())
+ .collect(ImmutableList.toImmutableList());
+ if (outputExprs.size() != 2 + outputColumns.size()
+ || !MergeOperation.OPERATION_COLUMN.equalsIgnoreCase(outputExprs.get(0).getName())
+ || !Column.ICEBERG_ROWID_COL.equalsIgnoreCase(outputExprs.get(1).getName())) {
+ throw new AnalysisException("Row-level merge sink output is not aligned with its routing columns "
+ + "and target columns");
}
- if (Column.ICEBERG_ROWID_COL.equalsIgnoreCase(name)) {
- return true;
+ List<NamedExpression> castExprs = Lists.newArrayListWithCapacity(outputExprs.size());
+ castExprs.add(outputExprs.get(0));
+ castExprs.add(outputExprs.get(1));
+ for (int i = 0; i < outputColumns.size(); ++i) {
+ Column column = outputColumns.get(i);
+ NamedExpression expr = outputExprs.get(i + 2);
+ castExprs.add(column.isVisible() ? coerceToColumn(expr, column) : expr);
+ }
+ return castExprs;
+ }
+
+ private static NamedExpression coerceToColumn(NamedExpression expr, Column column) {
+ DataType targetType = DataType.fromCatalogType(column.getType());
+ Expression castExpr = TypeCoercionUtils.castIfNotSameType(expr, targetType);
+ if (castExpr instanceof NamedExpression) {
+ NamedExpression namedExpr = (NamedExpression) castExpr;
+ if (column.getName().equalsIgnoreCase(namedExpr.getName())) {
+ return namedExpr;
+ }
+ return new Alias(namedExpr, column.getName());
}
- // Connector-reserved passthrough columns (iceberg v3 row-lineage), matched case-insensitively by name
- // — the names come from the sink's target Columns marked reservedPassthrough, so fe-core no longer
- // knows the iceberg column names. The reserved set is tiny (0-2), so a linear scan is fine.
- return reservedPassthroughNames.stream().anyMatch(n -> n.equalsIgnoreCase(name));
+ return new Alias(castExpr, column.getName());
}
private static boolean hasUnboundPlan(Plan plan) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
index 58d4d750910..9c294ec64a6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
@@ -40,6 +40,7 @@ import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.handle.ConnectorTableHandle;
import org.apache.doris.connector.spi.write.ConnectorChangelogMode;
import org.apache.doris.datasource.ExternalDatabase;
+import org.apache.doris.datasource.VariantWritePlanValidator;
import org.apache.doris.datasource.connector.converter.ConnectorWriteValueConverter;
import org.apache.doris.datasource.doris.RemoteDorisExternalTable;
import org.apache.doris.datasource.plugin.PluginDrivenExternalCatalog;
@@ -122,6 +123,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import java.util.stream.IntStream;
/**
* bind an unbound logicalTableSink represent the target table of an insert command
@@ -961,6 +963,11 @@ public class BindSink implements AnalysisRuleFactory {
throw new AnalysisException("Connector changelog sink must produce an operation column and "
+ targetWriteSchema.size() + " table columns, but got " + outputExpressions.size());
}
+ // The operation column precedes one column per table column.
+ List<Integer> tableColumnOrdinals = IntStream.rangeClosed(1, targetWriteSchema.size()).boxed()
+ .collect(ImmutableList.toImmutableList());
+ VariantWritePlanValidator.validateNoLossyCoercion(
+ targetWriteSchema, child, tableColumnOrdinals, ctx.cascadesContext.getCteContext());
return new LogicalConnectorTableSink<>(database, table, targetWriteSchema,
targetMetadata.getPartitionColumns(), targetMetadata.getWriteMetadataIdentity(),
targetWriteSchema, outputExpressions, sink.getDMLCommandType(), false,
@@ -998,6 +1005,7 @@ public class BindSink implements AnalysisRuleFactory {
throw new AnalysisException("insert into cols should be corresponding to the query output. "
+ "Expected " + boundSink.getCols().size() + " columns but got " + child.getOutput().size());
}
+ VariantWritePlanValidator.validateNoLossyCoercion(bindColumns, child, ctx.cascadesContext.getCteContext());
if (table.requiresFullSchemaWriteOrder()) {
// Positional-write connector (e.g. MaxCompute): its BE writer maps data columns positionally
// against the full table schema, so project the child to FULL-SCHEMA order with any
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/types/VariantType.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/types/VariantType.java
index 3e4d2103405..cc45189a076 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/types/VariantType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/types/VariantType.java
@@ -315,4 +315,20 @@ public class VariantType extends PrimitiveType {
|| dataType.isUuidType();
}
+ /** Whether the type or any nested complex type contains Variant. */
+ public static boolean containsVariant(DataType dataType) {
+ if (dataType instanceof VariantType) {
+ return true;
+ } else if (dataType instanceof ArrayType) {
+ return containsVariant(((ArrayType) dataType).getItemType());
+ } else if (dataType instanceof MapType) {
+ MapType mapType = (MapType) dataType;
+ return containsVariant(mapType.getKeyType()) || containsVariant(mapType.getValueType());
+ } else if (dataType instanceof StructType) {
+ return ((StructType) dataType).getFields().stream()
+ .anyMatch(field -> containsVariant(field.getDataType()));
+ }
+ return false;
+ }
+
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java
new file mode 100644
index 00000000000..863c5228258
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/VariantWritePlanValidatorTest.java
@@ -0,0 +1,625 @@
+// 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.
+
+package org.apache.doris.datasource;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.nereids.CTEContext;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CTEId;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.GreaterThan;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.ScalarSubquery;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.WhenClause;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.algebra.SetOperation.Qualifier;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCTEConsumer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRecursiveUnion;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRecursiveUnionAnchor;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRecursiveUnionProducer;
+import org.apache.doris.nereids.trees.plans.logical.LogicalSubQueryAlias;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWorkTableReference;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.ConnectorComputeVariantType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.function.Function;
+
+public class VariantWritePlanValidatorTest {
+
+ @Test
+ public void testUnionIfAndCaseImplicitVariantLossAreRejected() {
+ List<Plan> lossyPlans = ImmutableList.of(
+ lossyUnionPlan(),
+ projectExpression(variant -> new Cast(variant, IntegerType.INSTANCE)),
+ projectExpression(variant -> new If(
+ BooleanLiteral.TRUE,
+ new Cast(variant, IntegerType.INSTANCE),
+ new IntegerLiteral(1))),
+ projectExpression(variant -> new CaseWhen(
+ ImmutableList.of(new WhenClause(
+ BooleanLiteral.TRUE,
+ new Cast(variant, IntegerType.INSTANCE))),
+ new IntegerLiteral(1))));
+
+ for (Plan plan : lossyPlans) {
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), plan));
+ Assertions.assertTrue(exception.getMessage().contains(
+ "VARIANT write cannot safely convert input column 'payload'"), exception.getMessage());
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testAnalyzedUnionIfAndCaseObjectAndArrayPlansAreRejected() {
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ List<String> sqlStatements = ImmutableList.of(
+ "SELECT 1, parse_to_variant('{\"kind\":\"union\"}') "
+ + "UNION ALL SELECT 2, 1",
+ "SELECT 3, parse_to_variant('[\"union\", 3]') "
+ + "UNION ALL SELECT 4, 1",
+ "SELECT 3, IF(TRUE, parse_to_variant('{\"kind\":\"if\"}'), 1)",
+ "SELECT 4, IF(TRUE, parse_to_variant('[\"if\", 4]'), 1)",
+ "SELECT 5, CASE WHEN TRUE THEN parse_to_variant('{\"kind\":\"case\"}') "
+ + "ELSE 1 END",
+ "SELECT 6, CASE WHEN TRUE THEN parse_to_variant('[\"case\", 6]') "
+ + "ELSE 1 END");
+
+ for (String sql : sqlStatements) {
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(sql).getPlan();
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan),
+ sql);
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testAnalyzedExplicitVariantCastIsAccepted() {
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ Plan analyzedPlan = PlanChecker.from(connectContext)
+ .analyze("SELECT 1, CAST(parse_to_variant('{\"explicit\":true}') AS INT)")
+ .getPlan();
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan);
+ }
+
+ @Test
+ public void testAnalyzedCteProducerLossIsRejected() {
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(
+ "WITH source AS (SELECT CASE WHEN TRUE "
+ + "THEN parse_to_variant('{\"cte\":true}') ELSE 1 END AS payload) "
+ + "SELECT 1, payload FROM source").getPlan();
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan));
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ }
+
+ @Test
+ public void testAnalyzedRecursiveUnionAndGenerateLossAreRejected() {
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ List<String> sqlStatements = ImmutableList.of(
+ "WITH RECURSIVE source AS ("
+ + "SELECT IF(TRUE, parse_to_variant("
+ + "'{\"kind\":\"recursive\"}'), 1) AS payload "
+ + "UNION ALL SELECT CAST(1 AS DECIMAL(38, 9)) FROM source) "
+ + "SELECT 1, payload FROM source",
+ "SELECT 1, generated_payload FROM (SELECT 1 AS seed) source "
+ + "LATERAL VIEW explode(ARRAY(IF(TRUE, "
+ + "parse_to_variant('{\"kind\":\"generate\"}'), 1))) generated "
+ + "AS generated_payload");
+
+ for (String sql : sqlStatements) {
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(sql).getPlan();
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan),
+ sql);
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testRecursiveCteLossCarriedThroughAnotherColumnIsRejected() {
+ // The recursive term writes payload from the work table's a, so from the second row on payload holds
+ // what the anchor's a lost. MUTATION: tracing only the requested column of each term -> accepted.
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), recursiveCtePlan(0)));
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+
+ // The anchor's a never reaches payload when the recursive term writes payload from payload.
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), recursiveCtePlan(1));
+ }
+
+ @Test
+ public void testScalarSubqueryLossIsRejected() {
+ LogicalProject<?> lossyQueryPlan = projectExpression(variant -> new If(
+ BooleanLiteral.TRUE,
+ new Cast(variant, IntegerType.INSTANCE),
+ new IntegerLiteral(1)));
+ LogicalOneRowRelation lossyQuerySource = oneRow(
+ new Alias(new ScalarSubquery(lossyQueryPlan), "payload"));
+
+ ScalarSubquery coercedSubquery = (ScalarSubquery) new ScalarSubquery(variantInput())
+ .withTypeCoercion(IntegerType.INSTANCE);
+ LogicalOneRowRelation coercedSubquerySource = oneRow(
+ new Alias(coercedSubquery, "payload"));
+
+ for (Plan plan : ImmutableList.of(lossyQuerySource, coercedSubquerySource)) {
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), plan));
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testCorrelatedScalarSubqueryLossIsRejected() {
+ LogicalProject<?> lossyOuterSource = projectExpression(variant -> new If(
+ BooleanLiteral.TRUE,
+ new Cast(variant, IntegerType.INSTANCE),
+ new IntegerLiteral(1)));
+ Slot correlatedSlot = lossyOuterSource.getOutput().get(0);
+ LogicalOneRowRelation correlatedQuery = oneRow(
+ new Alias(correlatedSlot, "correlated_payload"));
+ ScalarSubquery correlatedSubquery = new ScalarSubquery(
+ correlatedQuery, ImmutableList.of(correlatedSlot), false);
+ LogicalProject<?> source = new LogicalProject<>(
+ ImmutableList.of(new Alias(correlatedSubquery, "payload")),
+ lossyOuterSource);
+
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), source));
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void testAnalyzerCteContextIsUsedBeforeAnchorConstruction() {
+ LogicalProject<?> lossyProducerChild = projectExpression(variant -> new If(
+ BooleanLiteral.TRUE,
+ new Cast(variant, IntegerType.INSTANCE),
+ new IntegerLiteral(1)));
+ LogicalSubQueryAlias<Plan> producerPlan = new LogicalSubQueryAlias<>(
+ "source", lossyProducerChild);
+ CTEId cteId = new CTEId(1);
+ CTEContext cteContext = new CTEContext(cteId, producerPlan, new CTEContext());
+ cteContext.setAnalyzedPlan(producerPlan);
+ LogicalCTEConsumer consumer = new LogicalCTEConsumer(
+ new RelationId(2), cteId, "source", producerPlan);
+ LogicalOneRowRelation source = oneRow(
+ new Alias(new ScalarSubquery(consumer), "payload"));
+
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), source, cteContext));
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void testWithNestedInAnalyzerContextProducerIsTraced() {
+ // A leading WITH binds its sink before the outer anchors exist, so the outer producer comes from the
+ // analyzer context and the WITH nested in it is not part of the traced plan. MUTATION: tracing the
+ // fetched producer without collecting its nested producers -> accepted.
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ Plan nestedWith = PlanChecker.from(connectContext).analyze(
+ "WITH inner_source AS (SELECT IF(TRUE, parse_to_variant('{\"lost\":true}'), 1) AS payload) "
+ + "SELECT payload FROM inner_source").getPlan();
+ LogicalSubQueryAlias<Plan> producerPlan = new LogicalSubQueryAlias<>("outer_source", nestedWith);
+ // Far from the ids the analyzer gave the nested WITH.
+ CTEId cteId = new CTEId(1000);
+ CTEContext cteContext = new CTEContext(cteId, producerPlan, new CTEContext());
+ cteContext.setAnalyzedPlan(producerPlan);
+ LogicalCTEConsumer consumer = new LogicalCTEConsumer(
+ new RelationId(1000), cteId, "outer_source", producerPlan);
+ LogicalOneRowRelation source = oneRow(
+ new Alias(new ScalarSubquery(consumer), "payload"));
+
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), source, cteContext));
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ }
+
+ @Test
+ public void testSharedCteProducerIsTracedOncePerRequestedOutputs() {
+ // Each CTE reads the previous one twice, so walking every path would trace c0 2^30 times. MUTATION:
+ // dropping the completed-trace cache -> times out.
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ StringBuilder sql = new StringBuilder("WITH c0 AS (SELECT parse_to_variant('{\"k\":1}') AS payload)");
+ for (int level = 1; level <= 30; level++) {
+ sql.append(String.format(", c%d AS (SELECT payload FROM c%d UNION ALL SELECT payload FROM c%d)",
+ level, level - 1, level - 1));
+ }
+ sql.append(" SELECT 1, payload FROM c30");
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(sql.toString()).getPlan();
+ Assertions.assertTimeoutPreemptively(Duration.ofSeconds(30),
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan));
+ }
+
+ @Test
+ public void testCteOutputsRequestedInManySetsAreTracedOnceEach() {
+ // c(i) reads c(i - 1) twice, for v0 alone and for v0 together with v(i), so the sets of outputs requested
+ // from c(24 - k) double with k: a cache keyed by the set of requested outputs would trace c0 2^24 times.
+ // Each output is traced once, and a loss carried by one of them only is still found. MUTATION: caching
+ // the traces per set of requested outputs -> times out or runs out of memory.
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ Plan lossless = PlanChecker.from(connectContext)
+ .analyze(cteChainRequestingDoublingSets(24, "parse_to_variant('{\"k\":1}')")).getPlan();
+ Assertions.assertTimeoutPreemptively(Duration.ofSeconds(30),
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), lossless));
+
+ Plan lossy = PlanChecker.from(connectContext).analyze(cteChainRequestingDoublingSets(
+ 24, "CAST(IF(TRUE, parse_to_variant('{\"lost\":true}'), 1) AS VARIANT)")).getPlan();
+ AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), lossy));
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ }
+
+ @Test
+ public void testWindowAndAggregateOrderingKeysAreNotWrittenValues() {
+ // The partition and order keys of a window and the ORDER BY of GROUP_CONCAT only arrange the rows whose
+ // values are used, so a lossy cast there, or in a column read only by such a key, writes nothing. A lossy
+ // cast in an argument of the function is written and still rejected. MUTATION: tracing the keys of a
+ // window, or the ORDER BY of an aggregate function -> rejected.
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ String source = "(SELECT 1 AS id, parse_to_variant('{\"k\":1}') AS v) source";
+ List<String> keyedOnly = ImmutableList.of(
+ "SELECT 1, ROW_NUMBER() OVER (ORDER BY IF(TRUE, v, 1)) FROM " + source,
+ "SELECT 1, COUNT(*) OVER (PARTITION BY IF(TRUE, v, 1)) FROM " + source,
+ "SELECT 1, GROUP_CONCAT('x' ORDER BY IF(TRUE, v, 1)) FROM " + source,
+ "SELECT 1, ROW_NUMBER() OVER (ORDER BY k) FROM (SELECT IF(TRUE, v, 1) AS k FROM " + source
+ + ") keyed");
+ for (String sql : keyedOnly) {
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(sql).getPlan();
+ Assertions.assertDoesNotThrow(
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan),
+ sql);
+ }
+
+ List<String> written = ImmutableList.of(
+ "SELECT 1, FIRST_VALUE(IF(TRUE, v, 1)) OVER (ORDER BY id) FROM " + source,
+ "SELECT 1, GROUP_CONCAT(CAST(IF(TRUE, v, 1) AS STRING) ORDER BY id) FROM " + source);
+ for (String sql : written) {
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(sql).getPlan();
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan),
+ sql);
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"),
+ exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testPrimitiveSourceAndExplicitVariantCastAreAccepted() {
+ LogicalOneRowRelation primitiveSource = oneRow(
+ new Alias(new IntegerLiteral(1), "payload"));
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), primitiveSource);
+
+ LogicalProject<?> explicitCastSource = projectExpression(
+ variant -> new Cast(variant, IntegerType.INSTANCE, true));
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), explicitCastSource);
+
+ LogicalOneRowRelation arrayVariantInput = oneRow(new Alias(
+ new NullLiteral(ArrayType.of(ConnectorComputeVariantType.INSTANCE)), "source_array"));
+ LogicalProject<?> arrayToVariantSource = new LogicalProject<>(ImmutableList.of(
+ new Alias(new Cast(
+ arrayVariantInput.getOutput().get(0),
+ ConnectorComputeVariantType.INSTANCE), "payload")),
+ arrayVariantInput);
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), arrayToVariantSource);
+ }
+
+ @Test
+ public void testLossyCastOutsideVariantTargetLineageIsIgnored() {
+ LogicalOneRowRelation input = variantInput();
+ Slot variantSlot = input.getOutput().get(0);
+ LogicalProject<?> source = new LogicalProject<>(ImmutableList.of(
+ new Alias(new Cast(variantSlot, IntegerType.INSTANCE), "id"),
+ new Alias(variantSlot, "payload")), input);
+ List<Column> targets = ImmutableList.of(
+ new Column("id", org.apache.doris.catalog.Type.INT),
+ new Column("payload", ConnectorComputeVariantType.INSTANCE.toCatalogDataType()));
+
+ VariantWritePlanValidator.validateNoLossyCoercion(targets, source);
+ }
+
+ @Test
+ public void testIfAndCaseConditionsAreNotWrittenValues() {
+ // The condition of an IF or of a CASE branch only chooses the value written: a Variant path compared
+ // with a scalar there is implicitly cast, but nothing written is lost. MUTATION: searching the whole
+ // expression tree for lossy casts -> rejected.
+ List<Plan> conditionCastPlans = ImmutableList.of(
+ projectExpression(variant -> new If(
+ new GreaterThan(new Cast(variant, IntegerType.INSTANCE), new IntegerLiteral(1)),
+ variant,
+ new NullLiteral(ConnectorComputeVariantType.INSTANCE))),
+ projectExpression(variant -> new CaseWhen(
+ ImmutableList.of(new WhenClause(
+ new GreaterThan(new Cast(variant, IntegerType.INSTANCE), new IntegerLiteral(1)),
+ variant)),
+ new NullLiteral(ConnectorComputeVariantType.INSTANCE))));
+ for (Plan plan : conditionCastPlans) {
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), plan);
+ }
+
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ List<String> sqlStatements = ImmutableList.of(
+ "SELECT 1, IF(payload['kind'] = 'event', payload, NULL) "
+ + "FROM (SELECT parse_to_variant('{\"kind\":\"event\"}') AS payload) source",
+ "SELECT 1, CASE WHEN payload['n'] > 1 THEN payload END "
+ + "FROM (SELECT parse_to_variant('{\"n\":2}') AS payload) source");
+ for (String sql : sqlStatements) {
+ Plan analyzedPlan = PlanChecker.from(connectContext).analyze(sql).getPlan();
+ Assertions.assertDoesNotThrow(
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), analyzedPlan),
+ sql);
+ }
+ }
+
+ @Test
+ public void testColumnsReadOnlyByConditionsAreNotTraced() {
+ // A column the target value reads only through a condition is not traced further: the lossy cast that
+ // computes it is not written. MUTATION: tracing every input slot of the target expression -> rejected.
+ LogicalOneRowRelation input = variantInput();
+ Slot variantSlot = input.getOutput().get(0);
+ LogicalProject<?> matched = new LogicalProject<>(ImmutableList.of(
+ new Alias(variantSlot, "payload"),
+ new Alias(new GreaterThan(new Cast(variantSlot, IntegerType.INSTANCE), new IntegerLiteral(1)),
+ "matched")), input);
+ LogicalProject<?> source = new LogicalProject<>(ImmutableList.of(
+ new Alias(new If(
+ matched.getOutput().get(1),
+ matched.getOutput().get(0),
+ new NullLiteral(ConnectorComputeVariantType.INSTANCE)), "payload")), matched);
+ VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), source);
+
+ // A lossy cast the branch value reads through a column is still rejected.
+ LogicalProject<?> lossyBranch = new LogicalProject<>(ImmutableList.of(
+ new Alias(new Cast(variantSlot, IntegerType.INSTANCE), "narrowed")), input);
+ LogicalProject<?> lossySource = new LogicalProject<>(ImmutableList.of(
+ new Alias(new If(
+ BooleanLiteral.TRUE,
+ lossyBranch.getOutput().get(0),
+ new IntegerLiteral(1)), "payload")), lossyBranch);
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), lossySource));
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void testLaterExceptAndIntersectInputsAreNotWrittenValues() {
+ // EXCEPT and INTERSECT write rows of their first input; the second input only decides which rows stay,
+ // so a lossy cast there writes nothing. MUTATION: tracing every set-operation input -> rejected.
+ ConnectContext connectContext = MemoTestUtils.createConnectContext();
+ String lossy = "IF(TRUE, parse_to_variant('{\"lost\":true}'), 1)";
+ String numeric = "CAST(1 AS DECIMAL(38, 9))";
+ for (String setOperator : ImmutableList.of("EXCEPT", "INTERSECT")) {
+ Plan membershipOnly = PlanChecker.from(connectContext).analyze(
+ "SELECT 1, " + numeric + " AS payload " + setOperator + " SELECT 1, " + lossy).getPlan();
+ Assertions.assertDoesNotThrow(
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), membershipOnly),
+ setOperator);
+
+ // The first input is written, so its lossy cast is still rejected.
+ Plan written = PlanChecker.from(connectContext).analyze(
+ "SELECT 1, " + lossy + " AS payload " + setOperator + " SELECT 1, " + numeric).getPlan();
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTargetWithId(), written),
+ setOperator);
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"),
+ exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testRowLevelDmlRoutingColumnsUseMappedSourceOrdinal() {
+ LogicalOneRowRelation input = variantInput();
+ Slot variantSlot = input.getOutput().get(0);
+ LogicalProject<?> source = new LogicalProject<>(ImmutableList.of(
+ new Alias(new IntegerLiteral(2), "operation"),
+ new Alias(new IntegerLiteral(7), "row_id"),
+ new Alias(new If(
+ BooleanLiteral.TRUE,
+ new Cast(variantSlot, IntegerType.INSTANCE),
+ new IntegerLiteral(1)), "payload")), input);
+
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(
+ variantTarget(), source, ImmutableList.of(2)));
+ Assertions.assertTrue(exception.getMessage().contains("input column 'payload'"), exception.getMessage());
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to INT"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void testDefaultMappingRequiresSameOutputWidth() {
+ LogicalOneRowRelation source = oneRow(
+ new Alias(new IntegerLiteral(2), "operation"),
+ new Alias(new IntegerLiteral(7), "row_id"),
+ new Alias(new NullLiteral(ConnectorComputeVariantType.INSTANCE), "payload"));
+
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(variantTarget(), source));
+ Assertions.assertTrue(exception.getMessage().contains("target and source columns are not aligned"),
+ exception.getMessage());
+ }
+
+ @Test
+ public void testLossyCastInNestedVariantTargetIsRejected() {
+ ArrayType sourceType = ArrayType.of(ConnectorComputeVariantType.INSTANCE);
+ LogicalOneRowRelation input = oneRow(new Alias(
+ new NullLiteral(sourceType), "source_variants"));
+ LogicalProject<?> source = new LogicalProject<>(ImmutableList.of(
+ new Alias(new Cast(
+ input.getOutput().get(0), ArrayType.of(IntegerType.INSTANCE)), "payloads")),
+ input);
+ List<Column> targets = ImmutableList.of(new Column(
+ "payloads", sourceType.toCatalogDataType()));
+
+ AnalysisException exception = Assertions.assertThrows(
+ AnalysisException.class,
+ () -> VariantWritePlanValidator.validateNoLossyCoercion(targets, source));
+ Assertions.assertTrue(exception.getMessage().contains("implicitly casts VARIANT to ARRAY<INT>"),
+ exception.getMessage());
+ }
+
+ /**
+ * {@code WITH RECURSIVE source(a, payload) AS (SELECT IF(TRUE, CAST(v AS INT), 1), 1 UNION ALL SELECT a, x
+ * FROM source) SELECT 1, payload FROM source}, shaped as AnalyzeCTE builds it, where x is the work-table
+ * column at {@code payloadSource}.
+ */
+ private static Plan recursiveCtePlan(int payloadSource) {
+ CTEId cteId = new CTEId(1);
+ LogicalOneRowRelation input = variantInput();
+ Plan anchor = new LogicalRecursiveUnionAnchor<>(cteId, new LogicalProject<>(ImmutableList.of(
+ new Alias(new If(
+ BooleanLiteral.TRUE,
+ new Cast(input.getOutput().get(0), IntegerType.INSTANCE),
+ new IntegerLiteral(1)), "a"),
+ new Alias(new IntegerLiteral(1), "payload")), input));
+ LogicalWorkTableReference workTable = new LogicalWorkTableReference(
+ new RelationId(2), cteId, anchor.getOutput(), ImmutableList.of("source"));
+ List<Slot> previous = workTable.getOutput();
+ Plan recursive = new LogicalRecursiveUnionProducer<>("source", new LogicalProject<>(ImmutableList.of(
+ new Alias(previous.get(0), "a"),
+ new Alias(previous.get(payloadSource), "payload")), workTable));
+ Plan union = new LogicalRecursiveUnion<>(
+ "source",
+ Qualifier.ALL,
+ ImmutableList.of(
+ new SlotReference("a", IntegerType.INSTANCE),
+ new SlotReference("payload", IntegerType.INSTANCE)),
+ ImmutableList.of(slotReferences(anchor), slotReferences(recursive)),
+ ImmutableList.of(anchor, recursive));
+ return new LogicalProject<>(ImmutableList.of(
+ new Alias(new IntegerLiteral(1), "id"), union.getOutput().get(1)), union);
+ }
+
+ /**
+ * WITH c0 (v0 ... v{levels}), then c(i) = SELECT COALESCE(v0, v(i)) AS v0, v1 ... FROM c(i - 1) UNION ALL
+ * SELECT v0, v1 ... FROM c(i - 1), and SELECT 1, v0 FROM c{levels}. The last column of c0 is {@code lastValue}
+ * and the others a Variant literal.
+ */
+ private static String cteChainRequestingDoublingSets(int levels, String lastValue) {
+ StringBuilder columns = new StringBuilder();
+ for (int column = 1; column <= levels; column++) {
+ columns.append(", v").append(column);
+ }
+ StringBuilder sql = new StringBuilder("WITH c0 AS (SELECT ");
+ for (int column = 0; column < levels; column++) {
+ sql.append("parse_to_variant('{\"k\":1}') AS v").append(column).append(", ");
+ }
+ sql.append(lastValue).append(" AS v").append(levels).append(')');
+ for (int level = 1; level <= levels; level++) {
+ sql.append(String.format(
+ ", c%d AS (SELECT COALESCE(v0, v%d) AS v0%s FROM c%d UNION ALL SELECT v0%s FROM c%d)",
+ level, level, columns, level - 1, columns, level - 1));
+ }
+ return sql.append(" SELECT 1, v0 FROM c").append(levels).toString();
+ }
+
+ private static List<SlotReference> slotReferences(Plan plan) {
+ return plan.getOutput().stream().map(SlotReference.class::cast).collect(ImmutableList.toImmutableList());
+ }
+
+ private static LogicalUnion lossyUnionPlan() {
+ LogicalProject<?> left = projectExpression(
+ variant -> new Cast(variant, IntegerType.INSTANCE));
+ LogicalOneRowRelation right = oneRow(new Alias(new IntegerLiteral(1), "payload"));
+ SlotReference output = new SlotReference("payload", IntegerType.INSTANCE);
+ return new LogicalUnion(
+ Qualifier.ALL,
+ ImmutableList.of(output),
+ ImmutableList.of(
+ ImmutableList.of((SlotReference) left.getOutput().get(0)),
+ ImmutableList.of((SlotReference) right.getOutput().get(0))),
+ ImmutableList.of(),
+ false,
+ ImmutableList.of(left, right));
+ }
+
+ private static LogicalProject<?> projectExpression(
+ Function<Slot, Expression> expressionFactory) {
+ LogicalOneRowRelation input = variantInput();
+ return new LogicalProject<>(
+ ImmutableList.of(new Alias(
+ expressionFactory.apply(input.getOutput().get(0)), "payload")),
+ input);
+ }
+
+ private static LogicalOneRowRelation variantInput() {
+ return oneRow(new Alias(
+ new NullLiteral(ConnectorComputeVariantType.INSTANCE), "source_variant"));
+ }
+
+ private static LogicalOneRowRelation oneRow(NamedExpression... expressions) {
+ return new LogicalOneRowRelation(
+ new RelationId(1), ImmutableList.copyOf(expressions));
+ }
+
+ private static List<Column> variantTarget() {
+ return ImmutableList.of(new Column(
+ "payload", ConnectorComputeVariantType.INSTANCE.toCatalogDataType()));
+ }
+
+ private static List<Column> variantTargetWithId() {
+ return ImmutableList.of(
+ new Column("id", org.apache.doris.catalog.Type.INT),
+ new Column("payload", ConnectorComputeVariantType.INSTANCE.toCatalogDataType()));
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindExpressionTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindExpressionTest.java
index 56ab6bdb7a8..5e76d20a3d4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindExpressionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/BindExpressionTest.java
@@ -17,24 +17,49 @@
package org.apache.doris.nereids.rules.analysis;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.common.NereidsException;
+import org.apache.doris.nereids.CTEContext;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.pattern.GeneratedPlanPatterns;
import org.apache.doris.nereids.rules.RulePromise;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Cast;
import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.functions.scalar.DayHourAdd;
import org.apache.doris.nereids.trees.expressions.functions.scalar.HourSecondSub;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
import org.apache.doris.nereids.trees.expressions.functions.scalar.MicroSecondsAdd;
import org.apache.doris.nereids.trees.expressions.functions.scalar.MicroSecondsSub;
import org.apache.doris.nereids.trees.expressions.functions.scalar.YearMonthAdd;
+import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.RelationId;
+import org.apache.doris.nereids.trees.plans.commands.merge.MergeOperation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalSort;
+import org.apache.doris.nereids.types.ConnectorComputeVariantType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.SmallIntType;
+import org.apache.doris.nereids.types.StringType;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.utframe.TestWithFeService;
+import com.google.common.collect.ImmutableList;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.List;
+
class BindExpressionTest extends TestWithFeService implements GeneratedPlanPatterns {
@Override
@@ -248,4 +273,44 @@ class BindExpressionTest extends TestWithFeService implements GeneratedPlanPatte
return RulePromise.REWRITE;
}
+ @Test
+ void testRowLevelMergeOutputIsCoercedByPosition() {
+ // A data column named like the operation routing column is matched by its position, so it is still cast to
+ // its target type. Matching by name skipped every cast of such an output.
+ List<NamedExpression> output = ImmutableList.of(
+ new Alias(new TinyIntLiteral((byte) 3), MergeOperation.OPERATION_COLUMN),
+ new SlotReference(Column.ICEBERG_ROWID_COL, StringType.INSTANCE),
+ new Alias(new IntegerLiteral(7), MergeOperation.OPERATION_COLUMN));
+ List<Column> columns = ImmutableList.of(
+ new Column(MergeOperation.OPERATION_COLUMN, PrimitiveType.SMALLINT));
+
+ List<NamedExpression> coerced = BindExpression.coerceMergeOutput(columns, output);
+
+ Assertions.assertEquals(3, coerced.size());
+ Assertions.assertEquals(SmallIntType.INSTANCE, coerced.get(2).getDataType());
+ Assertions.assertEquals(MergeOperation.OPERATION_COLUMN, coerced.get(2).getName());
+
+ AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+ () -> BindExpression.coerceMergeOutput(columns, output.subList(0, 2)));
+ Assertions.assertTrue(exception.getMessage().contains("not aligned"), exception.getMessage());
+ }
+
+ @Test
+ void testRowLevelMergeValidatesVariantTargetNamedLikeTheOperationColumn() {
+ // The Variant target named like the operation routing column is traced at its own position.
+ LogicalOneRowRelation input = new LogicalOneRowRelation(new RelationId(1), ImmutableList.of(
+ new Alias(new NullLiteral(ConnectorComputeVariantType.INSTANCE), "source_payload")));
+ Slot sourcePayload = input.getOutput().get(0);
+ LogicalProject<?> source = new LogicalProject<>(ImmutableList.of(
+ new Alias(new TinyIntLiteral((byte) 3), MergeOperation.OPERATION_COLUMN),
+ new Alias(new StringLiteral("row"), Column.ICEBERG_ROWID_COL),
+ new Alias(new If(BooleanLiteral.TRUE, new Cast(sourcePayload, IntegerType.INSTANCE),
+ new IntegerLiteral(1)), MergeOperation.OPERATION_COLUMN)), input);
+ List<Column> columns = ImmutableList.of(new Column(
+ MergeOperation.OPERATION_COLUMN, ConnectorComputeVariantType.INSTANCE.toCatalogDataType()));
+
+ AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+ () -> BindExpression.validateMergeNoLossyCoercion(columns, source, new CTEContext()));
+ Assertions.assertTrue(exception.getMessage().contains("input column 'operation'"), exception.getMessage());
+ }
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
index cae860ff253..e6287d502c2 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/ExternalRowLevelMergePlanBuilderTest.java
@@ -28,7 +28,9 @@ import org.apache.doris.nereids.analyzer.UnboundAlias;
import org.apache.doris.nereids.analyzer.UnboundSlot;
import org.apache.doris.nereids.trees.expressions.Cast;
import org.apache.doris.nereids.trees.expressions.EqualTo;
+import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.functions.scalar.ShortCircuitIf;
import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral;
import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
@@ -38,6 +40,8 @@ import org.apache.doris.nereids.trees.plans.commands.merge.MergeMatchedClause;
import org.apache.doris.nereids.trees.plans.logical.LogicalExternalRowLevelMergeSink;
import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.types.ConnectorComputeVariantType;
+import org.apache.doris.nereids.types.IntegerType;
import org.apache.doris.nereids.types.VarBinaryType;
import org.apache.doris.qe.ConnectContext;
@@ -140,4 +144,22 @@ public class ExternalRowLevelMergePlanBuilderTest {
Assertions.assertInstanceOf(ShortCircuitIf.class, projections.get(0).child(0),
"Iceberg MERGE must evaluate only the selected branch regardless of the session setting");
}
+
+ @Test
+ public void mergeBranchesAreCastToTheVariantTargetBeforeTheyAreSelected() {
+ // Each branch is cast to the Variant target before the branch selection, so selecting among a Variant and a
+ // numeric branch never derives a numeric common type that would turn an object or array Variant into NULL.
+ SlotReference targetPayload = new SlotReference("target_payload", ConnectorComputeVariantType.INSTANCE);
+ List<NamedExpression> projections = ExternalRowLevelMergePlanBuilder.generateFinalProjections(
+ ImmutableList.of("payload"), ImmutableList.of(ConnectorComputeVariantType.INSTANCE),
+ ImmutableList.of(ImmutableList.of(targetPayload), ImmutableList.of(new IntegerLiteral(1))));
+
+ Expression selection = projections.get(0).child(0);
+ Assertions.assertInstanceOf(ShortCircuitIf.class, selection);
+ Expression insertedBranch = selection.child(1);
+ Assertions.assertInstanceOf(Cast.class, insertedBranch);
+ Assertions.assertEquals(ConnectorComputeVariantType.INSTANCE, insertedBranch.getDataType());
+ Assertions.assertEquals(IntegerType.INSTANCE, insertedBranch.child(0).getDataType());
+ Assertions.assertEquals(ConnectorComputeVariantType.INSTANCE, selection.child(2).child(1).getDataType());
+ }
}
diff --git a/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_defaults.out b/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_defaults.out
new file mode 100644
index 00000000000..6e69c28cb0c
--- /dev/null
+++ b/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_defaults.out
@@ -0,0 +1,23 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !variant_default_historical_row --
+1 true \N
+
+-- !variant_default_rows --
+1 true \N
+2 true \N
+3 true \N
+4 true \N
+5 true \N
+6 true \N
+7 false null
+8 true \N
+
+-- !variant_default_counts --
+8 1 7
+
+-- !variant_explicit_default_null_column --
+8 0 8
+
+-- !variant_orc_historical_null --
+10 true \N
+
diff --git a/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out b/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
index 5427f2fb9d4..bc702cb62a5 100644
--- a/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
+++ b/regression-test/data/external_table_p0/iceberg/test_iceberg_variant_read.out
@@ -131,6 +131,9 @@ true 5 170 4.17
-- !variant_nested_projection --
1 first {"deep":{"name":"inside"},"x":11} {"kind":"open","score":101} {"enabled":true,"score":1001}
2 second \N null {"enabled":false,"score":2002}
+3 doris-nested {"kind":"struct","n":3} null \N
+4 empty {} \N \N
+5 \N \N \N \N
-- !variant_nested_filter --
1
@@ -141,6 +144,9 @@ true 5 170 4.17
-- !variant_nested_expressions --
1 11 inside open \N true 1002
2 \N \N \N 202 false 2003
+3 \N \N \N \N \N \N
+4 \N \N \N \N \N \N
+5 \N \N \N \N \N \N
-- !variant_evolution_renamed_snapshot --
1 initial 10 v1
diff --git a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
index f1b82f09fba..f2333c25116 100644
--- a/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
+++ b/regression-test/data/external_table_p0/paimon/write/test_paimon_write_variant_errors.out
@@ -11,3 +11,9 @@
28 false \N select-string
29 false \N false
30 false \N [3,4]
+
+-- !variant_dml_after_errors --
+1 {"kind":"updated"}
+2 ["array",2]
+3 {"kind":"object-source"}
+
diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_defaults.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_defaults.groovy
new file mode 100644
index 00000000000..93b367e7f6c
--- /dev/null
+++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_defaults.groovy
@@ -0,0 +1,156 @@
+// 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_variant_defaults",
+ "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_variant_defaults"
+ String dbName = "iceberg_variant_defaults_db"
+
+ spark_iceberg_multi """
+ CREATE NAMESPACE IF NOT EXISTS demo.${dbName};
+ DROP TABLE IF EXISTS demo.${dbName}.variant_defaults;
+ CREATE TABLE demo.${dbName}.variant_defaults (id INT) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read'
+ );
+ INSERT INTO demo.${dbName}.variant_defaults VALUES (1);
+ ALTER TABLE demo.${dbName}.variant_defaults ADD COLUMN payload VARIANT;
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_defaults_orc_history;
+ CREATE TABLE demo.${dbName}.variant_defaults_orc_history (id INT) USING iceberg
+ TBLPROPERTIES ('format-version'='3', 'write.format.default'='orc');
+ INSERT INTO demo.${dbName}.variant_defaults_orc_history VALUES (10);
+ ALTER TABLE demo.${dbName}.variant_defaults_orc_history 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"""
+
+ order_qt_variant_default_historical_row """
+ SELECT id, payload IS NULL, CAST(payload AS STRING)
+ FROM variant_defaults
+ ORDER BY id
+ """
+
+ sql """INSERT INTO variant_defaults (id) VALUES (2)"""
+ sql """INSERT INTO variant_defaults (id, payload) VALUES (3, DEFAULT)"""
+ sql """
+ INSERT INTO variant_defaults VALUES
+ (4, PARSE_TO_VARIANT('{"source":"update"}')),
+ (5, PARSE_TO_VARIANT('{"source":"merge"}'))
+ """
+ sql """UPDATE variant_defaults SET payload = DEFAULT(payload) WHERE id = 4"""
+ sql """
+ MERGE INTO variant_defaults t
+ USING (SELECT 5 AS id UNION ALL SELECT 6 AS id) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET payload = DEFAULT(payload)
+ WHEN NOT MATCHED THEN INSERT (id, payload) VALUES (s.id, DEFAULT(payload))
+ """
+ sql """INSERT INTO variant_defaults VALUES (7, PARSE_TO_VARIANT('null'))"""
+ sql """
+ INSERT INTO variant_defaults (id, payload)
+ SELECT 8, DEFAULT(payload) FROM variant_defaults WHERE id = 7
+ """
+
+ order_qt_variant_default_rows """
+ SELECT id, payload IS NULL, CAST(payload AS STRING)
+ FROM variant_defaults
+ ORDER BY id
+ """
+ qt_variant_default_counts """
+ SELECT COUNT(*), COUNT(payload), SUM(payload IS NULL)
+ FROM variant_defaults
+ """
+
+ sql """ALTER TABLE variant_defaults ADD COLUMN auxiliary VARIANT DEFAULT NULL"""
+ qt_variant_explicit_default_null_column """
+ SELECT COUNT(*), COUNT(auxiliary), SUM(auxiliary IS NULL)
+ FROM variant_defaults
+ """
+
+ List<List<Object>> schemaBeforeInvalidDefault = sql """DESC variant_defaults"""
+ test {
+ sql """
+ ALTER TABLE variant_defaults
+ ADD COLUMN invalid_payload VARIANT DEFAULT '{"invalid":true}'
+ """
+ exception "support null"
+ }
+ assertEquals(schemaBeforeInvalidDefault, sql("DESC variant_defaults"),
+ "Rejected non-null VARIANT default changed the Iceberg schema")
+
+ test {
+ sql """
+ ALTER TABLE variant_defaults
+ ADD COLUMN required_payload VARIANT NOT NULL DEFAULT NULL
+ """
+ exception "null default value"
+ }
+ assertEquals(schemaBeforeInvalidDefault, sql("DESC variant_defaults"),
+ "Rejected required VARIANT default changed the Iceberg schema")
+
+ // Spark caches Iceberg snapshots, so refresh after Doris commits before cross-engine reads.
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_defaults"""
+ List<List<Object>> sparkCounts = spark_iceberg """
+ SELECT COUNT(*), COUNT(payload), COUNT(auxiliary)
+ FROM demo.${dbName}.variant_defaults
+ """
+ assertEquals(1, sparkCounts.size())
+ assertEquals("8", sparkCounts[0][0].toString())
+ assertEquals("1", sparkCounts[0][1].toString())
+ assertEquals("0", sparkCounts[0][2].toString())
+
+ order_qt_variant_orc_historical_null """
+ SELECT id, payload IS NULL, CAST(payload AS STRING)
+ FROM variant_defaults_orc_history
+ ORDER BY id
+ """
+}
diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
index 7f7f29756e8..4452274410b 100644
--- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
+++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_variant_read.groovy
@@ -222,6 +222,18 @@ suite("test_iceberg_variant_read",
'{"n":', id, ',"padding":"', repeat('x', 256), '"}'))
FROM range(0, 8192);
+ DROP TABLE IF EXISTS demo.${dbName}.variant_doris_multi_row_group;
+ CREATE TABLE demo.${dbName}.variant_doris_multi_row_group (
+ id BIGINT,
+ info STRUCT<label: STRING, payload: VARIANT>,
+ padding STRING
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false'
+ );
+
DROP TABLE IF EXISTS demo.${dbName}.variant_deletion_vector;
CREATE TABLE demo.${dbName}.variant_deletion_vector (id INT, v VARIANT) USING iceberg
TBLPROPERTIES (
@@ -278,6 +290,21 @@ suite("test_iceberg_variant_read",
array(parse_json('null'), parse_json('{"kind":"close","score":202}')),
map('primary', parse_json('{"enabled":false,"score":2002}'));
+ DROP TABLE IF EXISTS demo.${dbName}.variant_nested_deep;
+ CREATE TABLE demo.${dbName}.variant_nested_deep (
+ id INT,
+ deep STRUCT<
+ level1: ARRAY<
+ MAP<STRING, STRUCT<note: STRING, payload: VARIANT>>
+ >
+ >
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false'
+ );
+
DROP TABLE IF EXISTS demo.${dbName}.variant_signed_selector;
CREATE TABLE demo.${dbName}.variant_signed_selector (
id INT,
@@ -306,6 +333,33 @@ suite("test_iceberg_variant_read",
CREATE TABLE demo.${dbName}.variant_write_guard (id INT) USING iceberg
TBLPROPERTIES ('format-version'='3', 'write.format.default'='parquet');
INSERT INTO demo.${dbName}.variant_write_guard VALUES (1);
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_coercion_guard;
+ CREATE TABLE demo.${dbName}.variant_coercion_guard (
+ id INT,
+ payload VARIANT
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false'
+ );
+
+ DROP TABLE IF EXISTS demo.${dbName}.variant_operation_column;
+ CREATE TABLE demo.${dbName}.variant_operation_column (
+ id INT,
+ `operation` VARIANT
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version'='3',
+ 'write.format.default'='parquet',
+ 'write.parquet.shred-variants'='false',
+ 'write.delete.mode'='merge-on-read',
+ 'write.update.mode'='merge-on-read',
+ 'write.merge.mode'='merge-on-read'
+ );
+ INSERT INTO demo.${dbName}.variant_operation_column
+ VALUES (1, parse_json('{"stage":"spark"}'));
"""
List<List<Object>> multiFileDataFiles = spark_iceberg """
@@ -459,6 +513,582 @@ public class AppendVariantEqualityDelete {
sql """set enable_profile=true"""
sql """set profile_level=2"""
+ // Doris currently writes only the unshredded metadata/value representation. Reject the
+ // table setting instead of silently ignoring the requested shredded encoding.
+ String shreddedWriteGuardSnapshot = latestSnapshotId("variant_page_pruning")
+ test {
+ sql """
+ INSERT INTO variant_page_pruning VALUES
+ (6000, PARSE_TO_VARIANT('{"shredding":"unsupported"}'))
+ """
+ exception "write.parquet.shred-variants=false"
+ }
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_page_pruning"""
+ assertEquals(shreddedWriteGuardSnapshot, latestSnapshotId("variant_page_pruning"),
+ "A rejected shredded Variant write must not commit a new snapshot")
+
+ sql """DROP TABLE IF EXISTS variant_doris_write_v2"""
+ test {
+ sql """
+ CREATE TABLE variant_doris_write_v2 (
+ id INT,
+ payload VARIANT
+ ) ENGINE=ICEBERG
+ PROPERTIES ('format-version'='2', 'write.format.default'='parquet')
+ """
+ exception "Iceberg VARIANT writes require table format-version 3"
+ }
+ sql """DROP TABLE IF EXISTS variant_doris_write_orc"""
+ test {
+ sql """
+ CREATE TABLE variant_doris_write_orc (
+ id INT,
+ payload VARIANT
+ ) ENGINE=ICEBERG
+ PROPERTIES ('format-version'='3', 'write.format.default'='orc')
+ """
+ exception "Iceberg VARIANT writes require Parquet data files"
+ }
+ sql """DROP TABLE IF EXISTS variant_doris_nested_ddl_guard"""
+ test {
+ sql """
+ CREATE TABLE variant_doris_nested_ddl_guard (
+ id INT,
+ info STRUCT<label: STRING, payload: VARIANT>
+ ) ENGINE=ICEBERG
+ PROPERTIES ('format-version'='3', 'write.format.default'='parquet')
+ """
+ exception "STRUCT unsupported sub-type: variant"
+ }
+
+ sql """DROP TABLE IF EXISTS variant_doris_write"""
+ sql """
+ CREATE TABLE variant_doris_write (
+ id INT,
+ payload VARIANT
+ ) ENGINE=ICEBERG
+ PROPERTIES ('format-version'='3', 'write.format.default'='parquet')
+ """
+ // Common-type analysis runs before sink binding. Reject implicit Variant-to-scalar casts
+ // instead of persisting an object/array as SQL NULL, while keeping ordinary scalar writes.
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT id, v FROM variant_values WHERE id = 1
+ UNION ALL
+ SELECT 2, 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT id + 2, v FROM variant_root_arrays WHERE id = 1
+ UNION ALL
+ SELECT 4, 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT 5, IF(TRUE, v, 1) FROM variant_values WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT 6, IF(TRUE, v, 1) FROM variant_root_arrays WHERE id = 2
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT 7, CASE WHEN TRUE THEN v ELSE 1 END
+ FROM variant_values WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT 8, CASE WHEN TRUE THEN v ELSE 1 END
+ FROM variant_root_arrays WHERE id = 2
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ WITH RECURSIVE source AS (
+ SELECT IF(TRUE, v, 1) AS payload FROM variant_values WHERE id = 1
+ UNION ALL
+ SELECT CAST(1 AS DECIMAL(38, 9)) FROM source
+ )
+ SELECT 9, payload FROM source
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO variant_coercion_guard
+ SELECT 9, generated_payload
+ FROM (SELECT v FROM variant_values WHERE id = 1) source
+ LATERAL VIEW explode(ARRAY(IF(TRUE, v, 1))) generated AS generated_payload
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+
+ // Primitive-to-Variant coercion remains supported for both inline VALUES and ordinary SELECT.
+ sql """
+ INSERT INTO variant_coercion_guard VALUES
+ (10, 7),
+ (11, 'values-string'),
+ (12, TRUE),
+ (13, ARRAY(1, 2))
+ """
+ sql """INSERT INTO variant_coercion_guard SELECT 20, 8"""
+ sql """INSERT INTO variant_coercion_guard SELECT 21, 'select-string'"""
+ sql """INSERT INTO variant_coercion_guard SELECT 22, FALSE"""
+ sql """INSERT INTO variant_coercion_guard SELECT 23, ARRAY(3, 4)"""
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_coercion_guard"""
+ List<List<Object>> sparkCoercionGuardRows = spark_iceberg """
+ SELECT id, to_json(payload)
+ FROM demo.${dbName}.variant_coercion_guard
+ ORDER BY id
+ """
+ assertEquals([
+ ["10", "7"],
+ ["11", '"values-string"'],
+ ["12", "true"],
+ ["13", "[1,2]"],
+ ["20", "8"],
+ ["21", '"select-string"'],
+ ["22", "false"],
+ ["23", "[3,4]"]
+ ], sparkCoercionGuardRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+
+ sql """
+ INSERT INTO variant_doris_write VALUES
+ (1, PARSE_TO_VARIANT('{"name":"doris","n":20,"enabled":true,"ratio":1.25,"nested":{"city":"hangzhou"},"tags":["iceberg","spark"]}')),
+ (2, PARSE_TO_VARIANT('{"name":"second","n":-7,"enabled":false,"ratio":-3.5,"nested":{"city":"shanghai"},"tags":["doris","variant"]}')),
+ (3, PARSE_TO_VARIANT('{"name":"unicode-中文","n":0,"enabled":true,"ratio":0.0,"nested":{"city":null},"tags":[]}')),
+ (4, PARSE_TO_VARIANT('[1,"two",null,{"k":4}]')),
+ (5, PARSE_TO_VARIANT('42')),
+ (6, PARSE_TO_VARIANT('"root-string"')),
+ (7, PARSE_TO_VARIANT('true')),
+ (8, PARSE_TO_VARIANT('null')),
+ (9, NULL),
+ (10, PARSE_TO_VARIANT('"null"')),
+ (11, PARSE_TO_VARIANT('""')),
+ (12, PARSE_TO_VARIANT('{}')),
+ (13, PARSE_TO_VARIANT('[]'))
+ """
+
+ // A single Doris writer emits more than 128 MiB of logical data, forcing multiple Parquet
+ // row groups in one file. The nested Variant leaf verifies that Iceberg metrics are merged
+ // across every row group instead of taking statistics from only the first one.
+ sql """
+ INSERT INTO variant_doris_multi_row_group
+ SELECT /*+ SET_VAR(parallel_pipeline_task_num=1) */
+ number,
+ NAMED_STRUCT(
+ 'label', 'doris-row-group',
+ 'payload', IF(
+ number % 10 = 0,
+ CAST(NULL AS VARIANT),
+ PARSE_TO_VARIANT(CONCAT('{"kind":"doris","n":', number, '}'))
+ )
+ ),
+ REPEAT('x', 2048)
+ FROM numbers("number" = "100000")
+ """
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_doris_multi_row_group"""
+ // Iceberg allocates IDs to all top-level fields before nested fields; payload is 5, not 4.
+ int nestedPayloadFieldId = 5
+ List<List<Object>> sparkDorisMultiRowGroupFiles = spark_iceberg """
+ SELECT file_format, record_count,
+ value_counts[${nestedPayloadFieldId}],
+ null_value_counts[${nestedPayloadFieldId}],
+ column_sizes[${nestedPayloadFieldId}] IS NULL,
+ lower_bounds[${nestedPayloadFieldId}] IS NULL,
+ upper_bounds[${nestedPayloadFieldId}] IS NULL
+ FROM demo.${dbName}.variant_doris_multi_row_group.files
+ WHERE content = 0
+ """
+ assertEquals(1, sparkDorisMultiRowGroupFiles.size(),
+ "The Doris multi-row-group write must produce exactly one data file")
+ assertEquals([
+ "parquet", "100000", "100000", "10000", "true", "true", "true"
+ ], sparkDorisMultiRowGroupFiles[0].collect { value ->
+ value == null ? null : value.toString().toLowerCase()
+ })
+ List<List<Object>> sparkDorisMultiRowGroupValues = spark_iceberg """
+ SELECT COUNT(*), COUNT(info.payload),
+ SUM(CASE WHEN info.payload IS NULL THEN 1 ELSE 0 END),
+ MIN(id), MAX(id), MIN(LENGTH(padding)), MAX(LENGTH(padding))
+ FROM demo.${dbName}.variant_doris_multi_row_group
+ """
+ assertEquals([
+ "100000", "90000", "10000", "0", "99999", "2048", "2048"
+ ], sparkDorisMultiRowGroupValues[0].collect { value -> value.toString() })
+ List<List<Object>> sparkDorisMultiRowGroupSamples = spark_iceberg """
+ SELECT id, info.label, variant_get(info.payload, '\$.n', 'bigint')
+ FROM demo.${dbName}.variant_doris_multi_row_group
+ WHERE id IN (0, 1, 99999)
+ ORDER BY id
+ """
+ assertEquals([
+ ["0", "doris-row-group", null],
+ ["1", "doris-row-group", "1"],
+ ["99999", "doris-row-group", "99999"]
+ ], sparkDorisMultiRowGroupSamples.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+ // CREATE/ADD followed by top-level MODIFY must remain metadata-only for Iceberg Variant.
+ sql """DROP TABLE IF EXISTS variant_doris_modify"""
+ sql """
+ CREATE TABLE variant_doris_modify (
+ id INT,
+ created_payload VARIANT NOT NULL
+ ) ENGINE=ICEBERG
+ PROPERTIES ('format-version'='3', 'write.format.default'='parquet')
+ """
+ sql """
+ ALTER TABLE variant_doris_modify
+ MODIFY COLUMN created_payload VARIANT NULL COMMENT 'created variant'
+ """
+ sql """ALTER TABLE variant_doris_modify ADD COLUMN added_payload VARIANT"""
+ sql """
+ ALTER TABLE variant_doris_modify
+ MODIFY COLUMN added_payload VARIANT COMMENT 'added variant' FIRST
+ """
+ sql """
+ INSERT INTO variant_doris_modify (id, created_payload, added_payload)
+ VALUES (1, PARSE_TO_VARIANT('{"source":"create"}'),
+ PARSE_TO_VARIANT('{"source":"add"}'))
+ """
+ List<List<Object>> sparkModifiedVariantRows = spark_iceberg """
+ SELECT id,
+ variant_get(created_payload, '\$.source', 'string'),
+ variant_get(added_payload, '\$.source', 'string')
+ FROM demo.${dbName}.variant_doris_modify
+ """
+ assertEquals([["1", "create", "add"]], sparkModifiedVariantRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+ List<List<Object>> sparkDorisVariantRows = spark_iceberg """
+ SELECT id,
+ variant_get(payload, '\$.name', 'string'),
+ variant_get(payload, '\$.n', 'int'),
+ variant_get(payload, '\$.enabled', 'boolean'),
+ variant_get(payload, '\$.ratio', 'double'),
+ variant_get(payload, '\$.nested.city', 'string'),
+ variant_get(payload, '\$.tags[1]', 'string'),
+ variant_get(payload, '\$[0]', 'int'),
+ variant_get(payload, '\$[1]', 'string'),
+ variant_get(payload, '\$[3].k', 'int'),
+ payload IS NULL
+ FROM demo.${dbName}.variant_doris_write
+ ORDER BY id
+ """
+ List<List<String>> sparkDorisVariantStrings = sparkDorisVariantRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString().toLowerCase() }
+ }
+ assertEquals([
+ ["1", "doris", "20", "true", "1.25", "hangzhou", "spark",
+ null, null, null, "false"],
+ ["2", "second", "-7", "false", "-3.5", "shanghai", "variant",
+ null, null, null, "false"],
+ ["3", "unicode-中文", "0", "true", "0.0", null, null,
+ null, null, null, "false"],
+ ["4", null, null, null, null, null, null, "1", "two", "4", "false"],
+ ["5", null, null, null, null, null, null, null, null, null, "false"],
+ ["6", null, null, null, null, null, null, null, null, null, "false"],
+ ["7", null, null, null, null, null, null, null, null, null, "false"],
+ ["8", null, null, null, null, null, null, null, null, null, "false"],
+ ["9", null, null, null, null, null, null, null, null, null, "true"],
+ ["10", null, null, null, null, null, null, null, null, null, "false"],
+ ["11", null, null, null, null, null, null, null, null, null, "false"],
+ ["12", null, null, null, null, null, null, null, null, null, "false"],
+ ["13", null, null, null, null, null, null, null, null, null, "false"]
+ ], sparkDorisVariantStrings)
+
+ // Keep SQL NULL, Variant null, the string "null", and empty values distinct.
+ List<List<Object>> sparkDorisRootRows = spark_iceberg """
+ SELECT id, to_json(payload), payload IS NULL
+ FROM demo.${dbName}.variant_doris_write
+ WHERE id BETWEEN 5 AND 13
+ ORDER BY id
+ """
+ assertEquals([
+ ["5", "42", "false"],
+ ["6", '"root-string"', "false"],
+ ["7", "true", "false"],
+ ["8", "null", "false"],
+ ["9", null, "true"],
+ ["10", '"null"', "false"],
+ ["11", '""', "false"],
+ ["12", "{}", "false"],
+ ["13", "[]", "false"]
+ ], sparkDorisRootRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+
+ List<List<Object>> sparkDorisVariantFiles = spark_iceberg """
+ SELECT file_format, record_count,
+ value_counts[2], null_value_counts[2],
+ lower_bounds[2] IS NULL, upper_bounds[2] IS NULL
+ FROM demo.${dbName}.variant_doris_write.files
+ WHERE content = 0
+ """
+ assertTrue(!sparkDorisVariantFiles.isEmpty())
+ assertTrue(sparkDorisVariantFiles.every { row -> row[0].toString().equalsIgnoreCase("parquet") })
+ assertEquals(13L, sparkDorisVariantFiles.sum { row -> Long.parseLong(row[1].toString()) })
+ assertEquals(13L, sparkDorisVariantFiles.sum { row -> Long.parseLong(row[2].toString()) })
+ assertEquals(1L, sparkDorisVariantFiles.sum { row -> Long.parseLong(row[3].toString()) })
+ assertTrue(sparkDorisVariantFiles.every { row -> row[4].toString().equalsIgnoreCase("true") })
+ assertTrue(sparkDorisVariantFiles.every { row -> row[5].toString().equalsIgnoreCase("true") })
+
+ // Doris writes Variant V2 leaves nested in every Iceberg complex container. Spark validates
+ // the committed Parquet Variant values, including SQL NULL versus Variant null and empties.
+ sql """
+ INSERT INTO variant_nested VALUES
+ (
+ 3,
+ NAMED_STRUCT(
+ 'label', 'doris-nested',
+ 'payload', PARSE_TO_VARIANT('{"kind":"struct","n":3}')
+ ),
+ ARRAY(
+ PARSE_TO_VARIANT('null'),
+ CAST(NULL AS VARIANT),
+ PARSE_TO_VARIANT('"null"'),
+ PARSE_TO_VARIANT('{}')
+ ),
+ MAP(
+ 'object', PARSE_TO_VARIANT('{"kind":"map","n":30}'),
+ 'json_null', PARSE_TO_VARIANT('null'),
+ 'sql_null', CAST(NULL AS VARIANT)
+ )
+ ),
+ (4, NAMED_STRUCT('label', 'empty', 'payload', PARSE_TO_VARIANT('{}')),
+ ARRAY(), MAP()),
+ (5, NULL, NULL, NULL)
+ """
+ // Spark caches Iceberg snapshots, so refresh after a Doris commit before cross-engine reads.
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_nested"""
+
+ List<List<Object>> sparkNestedContainers = spark_iceberg """
+ SELECT id,
+ info.label,
+ variant_get(info.payload, '\$.kind', 'string'),
+ variant_get(info.payload, '\$.n', 'int'),
+ info.payload IS NULL,
+ events IS NULL, size(events),
+ attrs IS NULL, size(attrs)
+ FROM demo.${dbName}.variant_nested
+ WHERE id IN (3, 4, 5)
+ ORDER BY id
+ """
+ assertEquals([
+ ["3", "doris-nested", "struct", "3", "false", "false", "4", "false", "3"],
+ ["4", "empty", null, null, "false", "false", "0", "false", "0"],
+ ["5", null, null, null, "true", "true", null, "true", null]
+ ], sparkNestedContainers.collect { row ->
+ row.collect { value -> value == null ? null : value.toString().toLowerCase() }
+ })
+
+ // Iceberg 1.10.1 cannot materialize Variant values from ArrayData/MapData in Spark 4.
+ // Doris still validates every nested value written to the shared Parquet files.
+ List<List<Object>> dorisNestedVariantValues = sql """
+ SELECT id,
+ CAST(info.payload['kind'] AS STRING),
+ CAST(info.payload['n'] AS INT),
+ VARIANT_TYPE(events[1]),
+ events[2] IS NULL,
+ VARIANT_TYPE(events[3]),
+ CAST(events[3] AS STRING),
+ VARIANT_TYPE(events[4]),
+ CAST(attrs['object']['kind'] AS STRING),
+ CAST(attrs['object']['n'] AS INT),
+ VARIANT_TYPE(attrs['json_null']),
+ attrs['sql_null'] IS NULL
+ FROM variant_nested
+ WHERE id = 3
+ """
+ assertEquals([[
+ "3", "struct", "3", "null", "true", "string", "null", "object",
+ "map", "30", "null", "true"
+ ]], dorisNestedVariantValues.collect { row ->
+ row.collect { value -> value == null ? null : value.toString().toLowerCase() }
+ })
+
+ sql """
+ INSERT INTO variant_nested_deep VALUES
+ (
+ 1,
+ NAMED_STRUCT(
+ 'level1',
+ ARRAY(
+ MAP(
+ 'outer',
+ NAMED_STRUCT(
+ 'note', 'from-doris',
+ 'payload', PARSE_TO_VARIANT(
+ '{"level2":{"level3":{"value":"deep-ok"}}}')
+ )
+ )
+ )
+ )
+ ),
+ (
+ 2,
+ NAMED_STRUCT(
+ 'level1',
+ ARRAY(MAP(
+ 'outer',
+ NAMED_STRUCT('note', 'sql-null', 'payload', CAST(NULL AS VARIANT))
+ ))
+ )
+ )
+ """
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_nested_deep"""
+ List<List<Object>> sparkDeepContainerRows = spark_iceberg """
+ SELECT id,
+ deep.level1[0]['outer'].note,
+ size(deep.level1)
+ FROM demo.${dbName}.variant_nested_deep
+ ORDER BY id
+ """
+ assertEquals([
+ ["1", "from-doris", "1"],
+ ["2", "sql-null", "1"]
+ ], sparkDeepContainerRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString().toLowerCase() }
+ })
+ List<List<Object>> dorisDeepVariantRows = sql """
+ SELECT id,
+ deep.level1[1]['outer'].note,
+ CAST(deep.level1[1]['outer'].payload['level2']['level3']['value'] AS STRING),
+ deep.level1[1]['outer'].payload IS NULL
+ FROM variant_nested_deep
+ ORDER BY id
+ """
+ assertEquals([
+ ["1", "from-doris", "deep-ok", "false"],
+ ["2", "sql-null", null, "true"]
+ ], dorisDeepVariantRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString().toLowerCase() }
+ })
+
+ // The first operation slot is merge-routing metadata; a quoted user column with the same
+ // name remains an ordinary data column and must still receive Variant V2 validation/coercion.
+ sql """
+ UPDATE variant_operation_column
+ SET `operation` = PARSE_TO_VARIANT('{"stage":"doris-update","n":1}')
+ WHERE id = 1
+ """
+ sql """
+ MERGE INTO variant_operation_column t
+ USING (
+ SELECT 1 AS id, PARSE_TO_VARIANT('{"stage":"doris-merge","n":2}') AS payload
+ UNION ALL
+ SELECT 2 AS id, PARSE_TO_VARIANT('{"stage":"doris-insert","n":3}') AS payload
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET `operation` = s.payload
+ WHEN NOT MATCHED THEN INSERT (id, `operation`) VALUES (s.id, s.payload)
+ """
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_operation_column"""
+ List<List<Object>> sparkOperationRows = spark_iceberg """
+ SELECT id,
+ variant_get(`operation`, '\$.stage', 'string'),
+ variant_get(`operation`, '\$.n', 'int')
+ FROM demo.${dbName}.variant_operation_column
+ ORDER BY id
+ """
+ assertEquals([
+ ["1", "doris-merge", "2"],
+ ["2", "doris-insert", "3"]
+ ], sparkOperationRows.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+
+ // Row-level DML must inspect the complete source lineage, not only the final sink-child type.
+ // Otherwise common-type coercion can turn an object Variant into SQL NULL before it is encoded.
+ test {
+ sql """
+ UPDATE variant_operation_column
+ SET `operation` = IF(TRUE, `operation`, 1)
+ WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column 'operation'"
+ }
+ test {
+ sql """
+ MERGE INTO variant_operation_column t
+ USING (
+ SELECT 1 AS id, PARSE_TO_VARIANT('{"stage":"lossy-merge"}') AS payload
+ UNION ALL
+ SELECT 2 AS id, 1 AS payload
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET `operation` = s.payload
+ """
+ exception "VARIANT write cannot safely convert input column 'operation'"
+ }
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_operation_column"""
+ List<List<Object>> sparkOperationRowsAfterRejectedDml = spark_iceberg """
+ SELECT id,
+ variant_get(`operation`, '\$.stage', 'string'),
+ variant_get(`operation`, '\$.n', 'int')
+ FROM demo.${dbName}.variant_operation_column
+ ORDER BY id
+ """
+ assertEquals([
+ ["1", "doris-merge", "2"],
+ ["2", "doris-insert", "3"]
+ ], sparkOperationRowsAfterRejectedDml.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+
+ // Coerce each MERGE action to the Iceberg target before building the action IF. Otherwise
+ // the unchanged object Variant and primitive insert value may first be coerced to DECIMAL,
+ // turning the object into SQL NULL before the sink sees it.
+ sql """
+ MERGE INTO variant_doris_write t
+ USING (
+ SELECT 1 AS id
+ UNION ALL
+ SELECT 14 AS id
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET payload = t.payload
+ WHEN NOT MATCHED THEN INSERT (id, payload) VALUES (s.id, 1)
+ """
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_doris_write"""
+ List<List<Object>> sparkUnchangedMergeVariant = spark_iceberg """
+ SELECT variant_get(payload, '\$.name', 'string'),
+ variant_get(payload, '\$.n', 'int'),
+ variant_get(payload, '\$.nested.city', 'string')
+ FROM demo.${dbName}.variant_doris_write
+ WHERE id = 1
+ """
+ assertEquals([["doris", "20", "hangzhou"]], sparkUnchangedMergeVariant.collect { row ->
+ row.collect { value -> value == null ? null : value.toString() }
+ })
+ List<List<Object>> sparkPrimitiveMergeVariant = spark_iceberg """
+ SELECT to_json(payload)
+ FROM demo.${dbName}.variant_doris_write
+ WHERE id = 14
+ """
+ assertEquals("1", sparkPrimitiveMergeVariant[0][0].toString())
+
def profileAction = new ProfileAction(context)
def mergedProfile = { String profile ->
if (!profile.contains("MergedProfile:")) {
@@ -747,6 +1377,21 @@ public class AppendVariantEqualityDelete {
FROM variant_type_matrix
"""
+ String dorisMultiRowGroupToken =
+ "iceberg_variant_doris_multi_row_group_" + UUID.randomUUID().toString()
+ List<List<Object>> dorisMultiRowGroupRows = sql """
+ SELECT '${dorisMultiRowGroupToken}', COUNT(*), SUM(LENGTH(padding)),
+ SUM(IF(info.payload IS NULL, 1, 0))
+ FROM variant_doris_multi_row_group
+ """
+ assertEquals([
+ dorisMultiRowGroupToken, "100000", "204800000", "10000"
+ ], dorisMultiRowGroupRows[0].collect { value -> value.toString() })
+ String dorisMultiRowGroupProfile = getProfileByToken(
+ dorisMultiRowGroupToken, ["RowGroupsTotalNum"]).toString()
+ assertTrue(counterSum(dorisMultiRowGroupProfile, "RowGroupsTotalNum") > 1,
+ "The single Doris-generated Parquet file did not contain multiple row groups")
+
String multiRowGroupColdToken =
"iceberg_variant_multi_row_group_cold_" + UUID.randomUUID().toString()
sql """
@@ -1027,17 +1672,21 @@ public class AppendVariantEqualityDelete {
exception "payload"
}
- test {
- sql """
- INSERT INTO variant_write_guard (id)
- SELECT id
- FROM variant_write_guard FOR VERSION AS OF ${writeGuardSourceSnapshot}
- """
- exception "Iceberg VARIANT columns are read-only and cannot be written"
- }
+ sql """
+ INSERT INTO variant_write_guard (id)
+ SELECT id
+ FROM variant_write_guard FOR VERSION AS OF ${writeGuardSourceSnapshot}
+ """
+ // Spark caches Iceberg snapshots, so refresh after Doris commits before cross-engine reads.
+ spark_iceberg """REFRESH TABLE demo.${dbName}.variant_write_guard"""
+ List<List<Object>> sparkPartialVariantRows = spark_iceberg """
+ SELECT COUNT(*), COUNT(payload)
+ FROM demo.${dbName}.variant_write_guard
+ """
+ assertEquals("2", sparkPartialVariantRows[0][0].toString())
+ assertEquals("0", sparkPartialVariantRows[0][1].toString())
- // A delete-only MERGE emits only position deletes. It must remain available even though
- // update/insert actions would route the unchanged Variant through the unsupported data writer.
+ // A delete-only MERGE still emits only position deletes and does not create a data file.
String beforePositionDeleteSnapshot = latestSnapshotId("variant_values")
sql """
MERGE INTO variant_values t
@@ -1072,9 +1721,16 @@ public class AppendVariantEqualityDelete {
// Files written before the Variant field existed have no physical Variant payload. Schema
// evolution must synthesize NULL instead of rejecting their non-Parquet file format.
+ // Metadata-only MODIFY is valid on an existing Iceberg v3 ORC Variant table. Doris still
+ // rejects data writes because Iceberg ORC has no Variant writer.
+ sql """ALTER TABLE variant_orc MODIFY COLUMN v VARIANT COMMENT 'orc payload'"""
order_qt_variant_orc_missing_column """
SELECT id, CAST(v AS STRING) FROM variant_orc ORDER BY id
"""
+ test {
+ sql """INSERT INTO variant_orc VALUES (2, CAST('{"format":"orc"}' AS VARIANT))"""
+ exception "Iceberg VARIANT writes require Parquet data files"
+ }
qt_variant_orc_count_star "SELECT COUNT(*) FROM variant_orc"
order_qt_variant_mixed_format """
SELECT id, CAST(v AS STRING) FROM variant_mixed_format ORDER BY id
diff --git a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
index 0ee47c37644..cde2e529c5e 100644
--- a/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
+++ b/regression-test/suites/external_table_p0/paimon/write/test_paimon_write_variant_errors.groovy
@@ -50,6 +50,17 @@ suite("test_paimon_write_variant_errors", "p0,external,paimon,nonConcurrent") {
payload VARIANT
) USING paimon
TBLPROPERTIES ('file.format' = 'parquet');
+
+ DROP TABLE IF EXISTS paimon.${dbName}.t_variant_dml_error;
+ CREATE TABLE paimon.${dbName}.t_variant_dml_error (
+ id INT,
+ payload VARIANT
+ ) USING paimon
+ TBLPROPERTIES (
+ 'file.format' = 'parquet',
+ 'primary-key' = 'id',
+ 'bucket' = '1'
+ );
"""
sql """DROP CATALOG IF EXISTS ${catalogName}"""
@@ -76,58 +87,81 @@ suite("test_paimon_write_variant_errors", "p0,external,paimon,nonConcurrent") {
(2, parse_to_variant('["array-source",2]'))
"""
- // Keep master's common-type semantics: mixed Variant and numeric expressions resolve to DECIMAL.
- // Object and array Variant values therefore become SQL NULL before the connector sink converts
- // the scalar result back to Paimon's Variant type.
- sql """
- INSERT INTO t_variant_error
- SELECT 100, payload FROM t_variant_coercion_source WHERE id = 1
- UNION ALL
- SELECT 101, 1
- """
- sql """
- INSERT INTO t_variant_error
- SELECT 102, payload FROM t_variant_coercion_source WHERE id = 2
- UNION ALL
- SELECT 103, 1
- """
- sql """
- INSERT INTO t_variant_error
- SELECT 104, IF(TRUE, payload, 1)
- FROM t_variant_coercion_source WHERE id = 1
- """
- sql """
- INSERT INTO t_variant_error
- SELECT 105, IF(TRUE, payload, 1)
- FROM t_variant_coercion_source WHERE id = 2
- """
- sql """
- INSERT INTO t_variant_error
- SELECT 106, CASE WHEN TRUE THEN payload ELSE 1 END
- FROM t_variant_coercion_source WHERE id = 1
- """
- sql """
- INSERT INTO t_variant_error
- SELECT 107, CASE WHEN TRUE THEN payload ELSE 1 END
- FROM t_variant_coercion_source WHERE id = 2
- """
- sql """
- INSERT INTO t_variant_error
- WITH RECURSIVE source(iter, payload) AS (
- SELECT CAST(0 AS INT), IF(TRUE, payload, 1)
- FROM t_variant_coercion_source WHERE id = 1
+ // The common type of a Variant and a number is DECIMAL. Writing it would turn an object or array Variant
+ // into SQL NULL before the sink converts the scalar back to Variant, so such writes are rejected.
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 100, payload FROM t_variant_coercion_source WHERE id = 1
UNION ALL
- SELECT CAST(iter + 1 AS INT), CAST(1 AS DECIMAL(38, 9))
- FROM source WHERE iter < 1
- )
- SELECT 108 + iter, payload FROM source
- """
- sql """
- INSERT INTO t_variant_error
- SELECT 110, generated_payload
- FROM (SELECT payload FROM t_variant_coercion_source WHERE id = 1) source
- LATERAL VIEW explode(ARRAY(IF(TRUE, payload, 1))) generated AS generated_payload
- """
+ SELECT 101, 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 102, payload FROM t_variant_coercion_source WHERE id = 2
+ UNION ALL
+ SELECT 103, 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 104, IF(TRUE, payload, 1)
+ FROM t_variant_coercion_source WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 105, IF(TRUE, payload, 1)
+ FROM t_variant_coercion_source WHERE id = 2
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 106, CASE WHEN TRUE THEN payload ELSE 1 END
+ FROM t_variant_coercion_source WHERE id = 1
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 107, CASE WHEN TRUE THEN payload ELSE 1 END
+ FROM t_variant_coercion_source WHERE id = 2
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ WITH RECURSIVE source(iter, payload) AS (
+ SELECT CAST(0 AS INT), IF(TRUE, payload, 1)
+ FROM t_variant_coercion_source WHERE id = 1
+ UNION ALL
+ SELECT CAST(iter + 1 AS INT), CAST(1 AS DECIMAL(38, 9))
+ FROM source WHERE iter < 1
+ )
+ SELECT 108 + iter, payload FROM source
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ INSERT INTO t_variant_error
+ SELECT 110, generated_payload
+ FROM (SELECT payload FROM t_variant_coercion_source WHERE id = 1) source
+ LATERAL VIEW explode(ARRAY(IF(TRUE, payload, 1))) generated AS generated_payload
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
sql """
INSERT INTO t_variant_error VALUES
@@ -168,28 +202,6 @@ suite("test_paimon_write_variant_errors", "p0,external,paimon,nonConcurrent") {
row.collect { value -> value == null ? null : value.toString() }
})
- def sparkCoercionRows = spark_paimon """
- SELECT id, to_json(payload)
- FROM paimon.${dbName}.t_variant_error
- WHERE id BETWEEN 100 AND 110
- ORDER BY id
- """
- assertEquals([
- ["100", null],
- ["101", "1"],
- ["102", null],
- ["103", "1"],
- ["104", null],
- ["105", null],
- ["106", null],
- ["107", null],
- ["108", null],
- ["109", "1"],
- ["110", null]
- ], sparkCoercionRows.collect { row ->
- row.collect { value -> value == null ? null : value.toString() }
- })
-
// Invalid JSON is preserved as a Variant string unless the global
// throw-on-invalid-JSON option is enabled.
order_qt_variant_after_errors """
@@ -197,9 +209,46 @@ suite("test_paimon_write_variant_errors", "p0,external,paimon,nonConcurrent") {
CAST(payload['recovered'] AS BOOLEAN),
payload
FROM t_variant_error
- WHERE id BETWEEN 20 AND 30
ORDER BY id
"""
+
+ // UPDATE and MERGE write their rows through the same check. A Variant compared with a scalar in the
+ // condition of an IF or a CASE only chooses the value written, so it is no lossy write.
+ sql """
+ INSERT INTO t_variant_dml_error VALUES
+ (1, parse_to_variant('{"kind":"object"}')),
+ (2, parse_to_variant('["array",2]'))
+ """
+ test {
+ sql """UPDATE t_variant_dml_error SET payload = IF(TRUE, payload, 1) WHERE id = 1"""
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ test {
+ sql """
+ MERGE INTO t_variant_dml_error t
+ USING (
+ SELECT 1 AS id, parse_to_variant('{"kind":"lossy-merge"}') AS payload
+ UNION ALL
+ SELECT 2 AS id, 1 AS payload
+ ) s
+ ON t.id = s.id
+ WHEN MATCHED THEN UPDATE SET payload = s.payload
+ """
+ exception "VARIANT write cannot safely convert input column 'payload'"
+ }
+ sql """
+ UPDATE t_variant_dml_error
+ SET payload = IF(payload['kind'] = 'object', parse_to_variant('{"kind":"updated"}'), payload)
+ WHERE id = 1
+ """
+ sql """
+ INSERT INTO t_variant_dml_error
+ SELECT 3, CASE WHEN payload['kind'] = 'object-source' THEN payload END
+ FROM t_variant_coercion_source WHERE id = 1
+ """
+ order_qt_variant_dml_after_errors """
+ SELECT id, CAST(payload AS STRING) FROM t_variant_dml_error ORDER BY id
+ """
} finally {
sql """SET force_jni_scanner = false"""
sql """DROP CATALOG IF EXISTS ${catalogName}"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org