eldenmoon commented on code in PR #66204:
URL: https://github.com/apache/doris/pull/66204#discussion_r3672093177
##########
be/src/storage/segment/variant/binary_column_extract_iterator.h:
##########
@@ -110,24 +116,119 @@ class BaseBinaryColumnProcessor : public ColumnIterator {
class BinaryColumnExtractIterator : public BaseBinaryColumnProcessor {
public:
BinaryColumnExtractIterator(std::string_view path, BinaryColumnCacheSPtr
sparse_column_cache,
- const StorageReadOptions* opts)
- : BaseBinaryColumnProcessor(std::move(sparse_column_cache), opts),
_path(path) {}
+ const StorageReadOptions* opts, bool
use_variant_v2)
+ : BaseBinaryColumnProcessor(std::move(sparse_column_cache), opts),
+ _path(path),
+ _use_variant_v2(use_variant_v2) {}
+
+ Status init(const ColumnIteratorOptions& opts) override {
+ if (_use_variant_v2) {
+ variant_v2::VariantAssemblerPlanOptions plan_options;
+ plan_options.mode =
variant_v2::VariantAssemblerMode::BINARY_EXTRACT;
+ std::shared_ptr<const variant_v2::VariantAssemblerPlan> plan;
+ RETURN_IF_ERROR(
+
variant_v2::VariantAssemblerPlan::create(std::move(plan_options), &plan));
+ _variant_v2_assembler =
std::make_unique<variant_v2::VariantAssembler>(std::move(plan));
+ }
+ return BaseBinaryColumnProcessor::init(opts);
+ }
// Batch processing using template method
Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
+ if (variant_v2::is_variant_v2_destination(*dst)) {
+ return _process_variant_v2_batch(
Review Comment:
[P1] 这里返回的是物理 Map reader 的 `has_null`，不是 assemble 后输出列的 null 状态。
持久化的 `Map<String,String>` 本身非空，但 `assemble_binary` 会把 requested path 缺失转换成
outer SQL NULL。新增的 `SharedCacheProducesTypedAndEncodedResults` 已经能证明这个不一致：`a` 的
row 1 是 NULL，而 `FixedSparseIterator` 明确写入 `has_null = false`，测试也没有检查返回
flag。因此这里可能返回 `has_null=false`，但 `dst` 实际包含 NULL。
建议把 flag 传入 V2 fill 路径，在 move `assembled` 前根据 `assembled.outer_nulls`
计算，并补对应断言。
##########
be/src/storage/segment/variant/variant_column_reader.cpp:
##########
@@ -417,12 +421,14 @@ Result<BinaryColumnCacheSPtr>
VariantColumnReader::_get_binary_column_cache(
}
DataTypePtr create_variant_type(const TabletColumn& target_col) {
- return target_col.is_nullable()
- ? make_nullable(std::make_shared<DataTypeVariant>(
- target_col.variant_max_subcolumns_count(),
- target_col.variant_enable_doc_mode()))
- :
std::make_shared<DataTypeVariant>(target_col.variant_max_subcolumns_count(),
-
target_col.variant_enable_doc_mode());
+ DataTypePtr type = target_col.variant_is_v2()
Review Comment:
[P3] 这里可以直接复用 `DataTypeFactory`，不要维护第二套 Variant type factory。
同一个 PR 已让 `DataTypeFactory::create_data_type(const TabletColumn&)` 支持
`variant_is_v2`、max subcolumns、doc mode 和 nullability；这个文件的其他 read-plan
分支也已经调用它。建议删除 `create_variant_type`，统一走 factory，避免后续增加 Variant property 或删除 V1
时需要同步两处。
##########
regression-test/suites/variant_p0/variant_compute_v2.groovy:
##########
@@ -18,8 +18,56 @@
suite("variant_compute_v2", "p0,nonConcurrent") {
sql "SET enable_nereids_planner = true"
sql "SET enable_fallback_to_original_planner = false"
+ def segmentScanTable = "variant_v2_segment_scan"
+ sql "DROP TABLE IF EXISTS ${segmentScanTable}"
+ sql """
+ CREATE TABLE ${segmentScanTable} (
+ id INT,
+ v VARIANT<PROPERTIES("variant_max_subcolumns_count" = "2")> NULL
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num" = "1")
+ """
+
+ // Storage writes remain on ColumnVariant. The same segment is scanned
into ColumnVariantV2
+ // after the session switch below.
+ sql "SET enable_variant_v2 = false"
Review Comment:
[P1] 新增的 enabled-INSERT guard 目前没有被测试到。
`LogicalOlapScan.createScanSlot` 的关键契约是：即使 `enable_variant_v2=true`，INSERT
内的 source scan 仍要产生 V1 slot，因为 writer 还是 V1。这里在 INSERT 前把开关设成 false，而且是
VALUES（没有 source `LogicalOlapScan`）；删掉 `StatementContext.isInsert()`
分支后这套测试仍然全绿。
建议补一个开关为 true 的 `INSERT ... SELECT`，并加 focused FE UT 覆盖 switch on/off ×
insert/select × whole/subpath slot。
##########
be/src/storage/segment/variant/hierarchical_data_iterator.cpp:
##########
@@ -107,36 +131,99 @@ Status
HierarchicalDataIterator::seek_to_ordinal(ordinal_t ord) {
DCHECK(_root_reader->inited);
RETURN_IF_ERROR(_root_reader->iterator->seek_to_ordinal(ord));
}
- DCHECK(_binary_column_reader->inited);
- RETURN_IF_ERROR(_binary_column_reader->iterator->seek_to_ordinal(ord));
+ if (_binary_column_reader) {
+ DCHECK(_binary_column_reader->inited);
+ RETURN_IF_ERROR(_binary_column_reader->iterator->seek_to_ordinal(ord));
+ }
return Status::OK();
}
Status HierarchicalDataIterator::next_batch(size_t* n, MutableColumnPtr& dst,
bool* has_null) {
- return process_read(
+ const size_t requested_rows = *n;
+ size_t actual_rows = 0;
+ RETURN_IF_ERROR(process_read(
[&](SubstreamIterator& reader, const PathInData& path, const
DataTypePtr& type) {
CHECK(reader.inited);
- RETURN_IF_ERROR(reader.iterator->next_batch(n, reader.column,
has_null));
- VLOG_DEBUG << fmt::format("{} next_batch {} rows, type={}",
path.get_path(), *n,
- type ? type->get_name() : "null");
- reader.rows_read += *n;
+ size_t stream_rows = requested_rows;
+ RETURN_IF_ERROR(reader.iterator->next_batch(&stream_rows,
reader.column, has_null));
Review Comment:
[P1] 不应让多个物理 stream 反复覆盖同一个输出 `has_null`。
当前顺序是 root -> materialized -> sparse；nullable root 报出的 `true` 可能被最后一个非
nullable sparse Map 覆盖成 `false`。反过来，materialized 子列的内部 missing 也可能报 `true`，但
outer Variant 行并不是 SQL NULL。V2 分支应给每个物理 reader 使用局部 flag，最终只根据 assembled outer
null map 设置调用方的 `has_null`。
##########
be/src/storage/segment/variant/binary_column_extract_iterator.h:
##########
@@ -110,24 +116,119 @@ class BaseBinaryColumnProcessor : public ColumnIterator {
class BinaryColumnExtractIterator : public BaseBinaryColumnProcessor {
public:
BinaryColumnExtractIterator(std::string_view path, BinaryColumnCacheSPtr
sparse_column_cache,
- const StorageReadOptions* opts)
- : BaseBinaryColumnProcessor(std::move(sparse_column_cache), opts),
_path(path) {}
+ const StorageReadOptions* opts, bool
use_variant_v2)
+ : BaseBinaryColumnProcessor(std::move(sparse_column_cache), opts),
+ _path(path),
+ _use_variant_v2(use_variant_v2) {}
+
+ Status init(const ColumnIteratorOptions& opts) override {
+ if (_use_variant_v2) {
+ variant_v2::VariantAssemblerPlanOptions plan_options;
+ plan_options.mode =
variant_v2::VariantAssemblerMode::BINARY_EXTRACT;
+ std::shared_ptr<const variant_v2::VariantAssemblerPlan> plan;
+ RETURN_IF_ERROR(
+
variant_v2::VariantAssemblerPlan::create(std::move(plan_options), &plan));
+ _variant_v2_assembler =
std::make_unique<variant_v2::VariantAssembler>(std::move(plan));
+ }
+ return BaseBinaryColumnProcessor::init(opts);
+ }
// Batch processing using template method
Status next_batch(size_t* n, MutableColumnPtr& dst, bool* has_null)
override {
+ if (variant_v2::is_variant_v2_destination(*dst)) {
+ return _process_variant_v2_batch(
+ [&]() { return _sparse_column_cache->next_batch(n,
has_null); }, n, dst);
+ }
return _process_batch([&]() { return
_sparse_column_cache->next_batch(n, has_null); }, *n,
dst);
}
// RowID-based read using template method
Status read_by_rowids(const rowid_t* rowids, const size_t count,
MutableColumnPtr& dst) override {
+ if (variant_v2::is_variant_v2_destination(*dst)) {
+ size_t rows = count;
+ return _process_variant_v2_batch(
+ [&]() { return
_sparse_column_cache->read_by_rowids(rowids, count); }, &rows,
+ dst);
+ }
return _process_batch([&]() { return
_sparse_column_cache->read_by_rowids(rowids, count); },
count, dst);
}
private:
std::string _path;
+ bool _use_variant_v2 = false;
+ std::unique_ptr<variant_v2::VariantAssembler> _variant_v2_assembler;
+
+ template <typename ReadMethod>
+ Status _process_variant_v2_batch(ReadMethod&& read_method, size_t*
num_rows,
+ MutableColumnPtr& dst) {
+ {
+
SCOPED_RAW_TIMER(&_read_opts->stats->variant_scan_sparse_column_timer_ns);
+ const int64_t before_size =
_read_opts->stats->uncompressed_bytes_read;
+ RETURN_IF_ERROR(read_method());
+ _read_opts->stats->variant_scan_sparse_column_bytes +=
+ _read_opts->stats->uncompressed_bytes_read - before_size;
+ }
+
SCOPED_RAW_TIMER(&_read_opts->stats->variant_fill_path_from_sparse_column_timer_ns);
+ if (_sparse_column_cache->binary_column->size() != *num_rows) {
+ return Status::Corruption("Variant sparse reader returned {} rows,
expected {}",
+
_sparse_column_cache->binary_column->size(), *num_rows);
+ }
+ return _fill_variant_v2_path(dst, *num_rows);
+ }
+
+ Status _fill_variant_v2_path(MutableColumnPtr& dst, size_t num_rows) {
+ DORIS_CHECK(_variant_v2_assembler != nullptr);
+ const auto* map =
+
check_and_get_column<ColumnMap>(_sparse_column_cache->binary_column.get());
+ if (map == nullptr || map->size() != num_rows) {
+ return Status::Corruption("Variant sparse input must be a {}-row
Map<String,String>",
+ num_rows);
+ }
+ const auto* paths =
check_and_get_column<ColumnString>(&map->get_keys());
+ const auto* values =
check_and_get_column<ColumnString>(&map->get_values());
+ if (paths == nullptr || values == nullptr || paths->size() !=
values->size()) {
+ return Status::Corruption("Variant sparse input is not
Map<String,String>");
+ }
+
+ const auto& offsets = map->get_offsets();
+ const StringRef requested {_path.data(), _path.size()};
+ DorisVector<StringRef> cells;
+ cells.reserve(num_rows);
+ DorisVector<uint8_t> missing;
+ missing.reserve(num_rows);
+ size_t previous_end = 0;
+ for (size_t row = 0; row < num_rows; ++row) {
+ const size_t end = offsets[ssize_t(row)];
+ if (end < previous_end || end > paths->size()) {
+ return Status::Corruption("Variant sparse row {} has invalid
offset {}", row, end);
+ }
+ const size_t lower =
ColumnVariant::find_path_lower_bound_in_sparse_data(
Review Comment:
[P2] 请把这个稳定的 sparse-storage primitive 从 `ColumnVariant` 类迁到中立层。
这里和 `variant_assembler.cpp:86` 都让新 V2 reader 反向依赖未来要删除的 V1 类。这个函数只是对排序后的
`ColumnString` 行区间做 lower_bound，与 V1 对象模型无关。建议抽到 neutral Variant storage/path
utility，让 V1/V2 都复用；过渡期可以在 V1 保留 forwarding wrapper。这样既不复制算法，也不会让后续删除 V1 时再修改
V2 reader。
##########
be/src/storage/segment/variant/v2/variant_assembler.h:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <span>
+
+#include "common/status.h"
+#include "core/column/column.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type.h"
+#include "core/types.h"
+#include "util/json/path_in_data.h"
+
+namespace doris {
+
+class ColumnMap;
+
+namespace segment_v2::variant_v2 {
+
+enum class VariantAssemblerMode : uint8_t {
+ HIERARCHICAL,
+ BINARY_EXTRACT,
+};
+
+struct VariantAssemblerMaterializedPath {
+ PathInData path;
+ DataTypePtr type;
+};
+
+struct VariantAssemblerPlanOptions {
+ VariantAssemblerMode mode = VariantAssemblerMode::HIERARCHICAL;
+ PathInData requested_path;
+ DorisVector<VariantAssemblerMaterializedPath> materialized_paths;
+ size_t sparse_bucket_count = 0;
Review Comment:
[P2] API 应先表达生产环境实际存在的 0/1 个 sparse stream。
唯一 hierarchical producer 只会传 0 或 1；物理 sparse buckets 在更上层已经由
`CombineMultipleBinaryColumnIterator` 合并。这里允许任意 count 和
`span<ColumnMap*>`，进一步驱动 `variant_assembler.cpp:513-580` 的 N-way
cursor/selection merge；目前唯一 `>1` caller 只是人工 corruption UT。
建议现在使用 optional/single sparse Map，等真正有生产 caller 传独立 buckets 时再泛化。
##########
be/src/storage/segment/variant/v2/variant_assembler_value.cpp:
##########
@@ -0,0 +1,1225 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <array>
+#include <cstring>
+#include <limits>
+#include <utility>
+
+#include "common/check.h"
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2_typed_column.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_nullable.h"
+#include "core/typeid_cast.h"
+#include "core/value/variant/variant_parquet_encoding.h"
+#include "exec/common/format_ip.h"
+#include "exprs/function/parse/variant_jsonb_parse.h"
+#include "storage/segment/variant/v2/variant_assembler_internal.h"
+#include "util/utf8_check.h"
+
+namespace doris::segment_v2::variant_v2::variant_assembler_internal {
+namespace {
+
+class CellCursor {
+public:
+ explicit CellCursor(StringRef cell)
+ : _current(reinterpret_cast<const uint8_t*>(cell.data)),
_remaining(cell.size) {
+ if (cell.data == nullptr && cell.size != 0) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage cell has a null pointer for {}
bytes", cell.size);
+ }
+ }
+
+ template <typename T>
+ T read(std::string_view description) {
+ require(sizeof(T), description);
+ T value;
+ std::memcpy(&value, _current, sizeof(T));
+ _current += sizeof(T);
+ _remaining -= sizeof(T);
+ return value;
+ }
+
+ StringRef read_bytes(size_t size, std::string_view description) {
+ require(size, description);
+ const StringRef result {reinterpret_cast<const char*>(_current), size};
+ _current += size;
+ _remaining -= size;
+ return result;
+ }
+
+ size_t remaining() const noexcept { return _remaining; }
+ bool empty() const noexcept { return _remaining == 0; }
+
+private:
+ void require(size_t size, std::string_view description) const {
+ if (size > _remaining) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Truncated Variant storage cell while reading {}:
need {} bytes, "
+ "have {}",
+ description, size, _remaining);
+ }
+ }
+
+ const uint8_t* _current;
+ size_t _remaining;
+};
+
+template <typename Integer>
+uint32_t decimal_digits(Integer value) {
+ uint32_t result = 0;
+ do {
+ value /= 10;
+ ++result;
+ } while (value != 0);
+ return result;
+}
+
+template <typename Integer>
+void validate_decimal(Integer value, uint8_t precision, uint8_t scale, uint8_t
maximum,
+ std::string_view description) {
+ if (precision == 0 || precision > maximum || scale > precision) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage {} precision/scale {}/{} is invalid",
description,
+ precision, scale);
+ }
+ if (decimal_digits(value) > precision) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage {} value exceeds declared precision
{}", description,
+ precision);
+ }
+}
+
+void validate_depth(uint32_t depth) {
+ if (depth > VARIANT_MAX_NESTING_DEPTH) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage cell exceeds maximum nesting depth
{}",
+ VARIANT_MAX_NESTING_DEPTH);
+ }
+}
+
+void validate_container_depth(uint32_t depth) {
+ if (depth >= VARIANT_MAX_NESTING_DEPTH) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage container exceeds maximum nesting
depth {}",
+ VARIANT_MAX_NESTING_DEPTH);
+ }
+}
+
+template <typename DateValue>
+int32_t storage_date_days(DateValue value, std::string_view description) {
+ if (!value.is_valid_date()) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid {} in Variant storage
cell", description);
+ }
+ return variant_days_since_epoch(value, 0, description);
+}
+
+template <typename DateTimeValue>
+int64_t storage_timestamp_micros(DateTimeValue value, std::string_view
description) {
+ if (!value.is_valid_date()) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid {} in Variant storage
cell", description);
+ }
+ return variant_timestamp_micros(value, 0, description);
+}
+
+VecDateTimeValue read_legacy_temporal(CellCursor& cursor, FieldType type) {
+ const bool is_date = type == FieldType::OLAP_FIELD_TYPE_DATE;
+ DORIS_CHECK(is_date || type == FieldType::OLAP_FIELD_TYPE_DATETIME);
+ const auto value = cursor.read<VecDateTimeValue>(is_date ? "legacy DATE" :
"legacy DATETIME");
+ if (is_date) {
+ static_cast<void>(storage_date_days(value, "legacy DATE"));
+ } else {
+ static_cast<void>(storage_timestamp_micros(value, "legacy DATETIME"));
+ }
+ return value;
+}
+
+struct LegacyDecimalCell {
+ uint8_t precision;
+ uint8_t scale;
+ __int128 value;
+};
+
+LegacyDecimalCell read_legacy_decimal(CellCursor& cursor) {
+ LegacyDecimalCell result {
+ .precision = cursor.read<uint8_t>("legacy DecimalV2 precision"),
+ .scale = cursor.read<uint8_t>("legacy DecimalV2 scale"),
+ .value = cursor.read<__int128>("legacy DecimalV2 value"),
+ };
+ if (result.precision == 0 || result.precision > DecimalV2Value::PRECISION
||
+ result.scale > result.precision) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage legacy DecimalV2 precision/scale
{}/{} is invalid",
+ result.precision, result.scale);
+ }
+ // DecimalV2 always stores a fixed nine-digit fractional coefficient.
Preserve the serialized
+ // SerDe metadata for typed identity, but do not use it to rescale the
coefficient.
+ validate_decimal(result.value, DecimalV2Value::PRECISION,
DecimalV2Value::SCALE,
+ DecimalV2Value::PRECISION, "legacy DecimalV2");
+ return result;
+}
+
+void add_largeint(VariantBatchBuilder::Row& output, __int128 value) {
+ output.add_largeint(value);
+}
+
+void add_decimal256(VariantBatchBuilder::Row& output, wide::Int256 value,
uint8_t scale) {
+ const std::string text = Decimal256 {value}.to_string(scale);
+ output.add_string(StringRef(text));
+}
+
+void add_ipv4(VariantBatchBuilder::Row& output, IPv4 value) {
+ std::array<char, IPV4_MAX_TEXT_LENGTH + 1> buffer {};
+ char* end = buffer.data();
+ const auto* address = reinterpret_cast<const unsigned char*>(&value);
+ format_ipv4(address, end);
+ output.add_string({buffer.data(), static_cast<size_t>(end -
buffer.data())});
+}
+
+void add_ipv6(VariantBatchBuilder::Row& output, IPv6 value) {
+ std::array<char, IPV6_MAX_TEXT_LENGTH + 1> buffer {};
+ char* end = buffer.data();
+ format_ipv6(reinterpret_cast<unsigned char*>(&value), end);
+ output.add_string({buffer.data(), static_cast<size_t>(end -
buffer.data())});
+}
+
+StringRef read_storage_string(CellCursor& cursor) {
+ const size_t size = cursor.read<size_t>("string size");
+ const StringRef value = cursor.read_bytes(size, "string payload");
+ if (value.size != 0 && !validate_utf8(value.data, value.size)) {
+ throw Exception(ErrorCode::CORRUPTION, "Variant storage string is not
valid UTF-8");
+ }
+ return value;
+}
+
+// Keep the exhaustive storage FieldType decoder together so validation and
byte consumption for
+// every wire tag remain auditable in one dispatch table.
+// NOLINTNEXTLINE(readability-function-size)
+void decode_storage_value(CellCursor& cursor, VariantBatchBuilder::Row&
output, uint32_t depth) {
Review Comment:
[P2] 持久化 storage-cell wire format 应只有一个 owner。
这个文件新增了 `decode_storage_value`、`inspect_typed_cell`、`append_cell_to_typed`
三套对相同 `FieldType` bytes 的解释；writer 走 nested
`DataTypeSerDe::write_one_cell_to_binary`，core 里也已有
`deserialize_binary_to_field/column`。以后增加类型或修改格式时，这几张表很容易在支持范围、边界检查或消费字节数上漂移。
建议在 `core/data_type_serde` 抽取/扩展一个 bounded、返回 `Status` 的 codec/visitor，补齐
legacy DATE/DATETIME/DECIMAL 和 trailing-byte 校验；assembler 只负责把 decoded value 写入
`VariantBatchBuilder`。这个中立 codec 在 V1 删除后仍然成立。
##########
regression-test/suites/variant_p0/variant_compute_v2.groovy:
##########
@@ -18,8 +18,56 @@
suite("variant_compute_v2", "p0,nonConcurrent") {
sql "SET enable_nereids_planner = true"
sql "SET enable_fallback_to_original_planner = false"
+ def segmentScanTable = "variant_v2_segment_scan"
+ sql "DROP TABLE IF EXISTS ${segmentScanTable}"
+ sql """
+ CREATE TABLE ${segmentScanTable} (
+ id INT,
+ v VARIANT<PROPERTIES("variant_max_subcolumns_count" = "2")> NULL
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num" = "1")
+ """
+
+ // Storage writes remain on ColumnVariant. The same segment is scanned
into ColumnVariantV2
+ // after the session switch below.
+ sql "SET enable_variant_v2 = false"
+ sql """
+ INSERT INTO ${segmentScanTable} VALUES
+ (0, '{"dense":1,"nested":{"value":"a"},"sparse":100}'),
+ (1, '{"dense":2,"nested":{"value":"b"}}'),
+ (2, '{"dense":3,"nested":{"value":"c"},"sparse":300}'),
+ (3, NULL)
+ """
sql "SET enable_variant_v2 = true"
+ order_qt_segment_scan """
+ SELECT id, CAST(v AS STRING), v IS NULL
+ FROM ${segmentScanTable}
+ ORDER BY id
+ """
+
+ order_qt_segment_scan_subpaths """
+ SELECT id,
+ CAST(v['dense'] AS BIGINT),
+ CAST(v['nested']['value'] AS STRING),
+ CAST(v['sparse'] AS BIGINT)
+ FROM ${segmentScanTable}
+ ORDER BY id
+ """
+
+ // variant_type accepts legacy ColumnVariant, but deliberately rejects
ColumnVariantV2.
Review Comment:
[P2] 不要用函数暂未支持 V2 作为 routing oracle。
这个 case 只有在 `variant_type` 故意不支持 ColumnVariantV2 时才通过。将来补齐函数或删除 V1，即使
storage routing 完全正确也会失败。建议在 FE 边界断言 scan slot/type，或通过稳定的 BE test hook 断言
iterator 输出类；SQL regression 只验证已支持的外部行为。
##########
be/test/storage/variant/variant_column_writer_reader_test.cpp:
##########
@@ -948,6 +971,435 @@ TEST_F(VariantColumnWriterReaderTest, test_statics) {
// EXPECT_EQ(stats.sparse_column_non_null_size["key7"], 300);
}
+TEST_F(VariantColumnWriterReaderTest, test_segment_rowid_read_variant_v2) {
+ init_variant_tablet(21000, 1);
+ // "hot" is present in every row and consumes the only materialized slot.
"z" remains in the
+ // shared sparse column, including one physically missing row.
+ const std::vector<std::string> jsons {R"({"hot":1,"z":101})",
R"({"hot":2})",
+ R"({"hot":3,"z":103})",
R"({"hot":4,"z":104})"};
+ auto rowset = create_variant_rowset({jsons}, 1, 100);
Review Comment:
[P2] 迁移契约需要冻结的历史 V1 bytes，不能全部由当前 writer 现写现读。
新增 segment/rowid cases 都用当前 V1 writer 生成数据后立刻交给新 reader；writer/reader
同步漂移时测试仍会绿，而且以后删掉 V1 writer，这些 fixture 也会一起消失。
建议增加一个由已发布版本生成、只读且有版本号的 legacy segment fixture，至少覆盖 root-only top-level
scalar/array/JSON null/SQL NULL，以及 materialized/sparse/doc，分别走 sequential 和
rowid。现有 round-trip 可以保留作为补充。
##########
be/src/storage/segment/variant/v2/variant_assembler.h:
##########
@@ -0,0 +1,118 @@
+// 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.
+
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <span>
+
+#include "common/status.h"
+#include "core/column/column.h"
+#include "core/column/variant_v2/column_variant_v2.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/data_type.h"
+#include "core/types.h"
+#include "util/json/path_in_data.h"
+
+namespace doris {
+
+class ColumnMap;
+
+namespace segment_v2::variant_v2 {
+
+enum class VariantAssemblerMode : uint8_t {
+ HIERARCHICAL,
+ BINARY_EXTRACT,
+};
+
+struct VariantAssemblerMaterializedPath {
+ PathInData path;
+ DataTypePtr type;
+};
+
+struct VariantAssemblerPlanOptions {
+ VariantAssemblerMode mode = VariantAssemblerMode::HIERARCHICAL;
+ PathInData requested_path;
+ DorisVector<VariantAssemblerMaterializedPath> materialized_paths;
+ size_t sparse_bucket_count = 0;
+ bool has_root = false;
+ bool has_doc = false;
+};
+
+// Immutable cold-path plan. Paths and physical types are validated and
retained so one plan can
Review Comment:
[P2] 当前没有生产调用方需要独立的 Plan 生命周期，建议收窄。
两处生产 callsite 都是 create plan 后立即 move 给唯一一个 assembler，没有 plan sharing。现在的
`shared_ptr<const Plan>`、Plan PImpl、Assembler PImpl、以及保留 first error 的 terminal
state 增加了两层 heap lifetime 和一个没有调用方需求的状态契约。真实 reader 在 `RETURN_IF_ERROR`
后会结束；`assemble` 内局部 `result` 已经保证失败时不发布半成品。
建议改成单一 `VariantAssembler::create(options)`，由 assembler value-own compiled
paths，并直接返回每次调用的错误。这样仍保留 cold-path compile，但 API/state 小很多。
##########
be/src/storage/segment/variant/v2/variant_assembler_value.cpp:
##########
@@ -0,0 +1,1225 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <array>
+#include <cstring>
+#include <limits>
+#include <utility>
+
+#include "common/check.h"
+#include "common/exception.h"
+#include "core/assert_cast.h"
+#include "core/column/column_decimal.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_vector.h"
+#include "core/column/variant_v2/column_variant_v2_typed_column.h"
+#include "core/data_type/data_type_array.h"
+#include "core/data_type/data_type_factory.hpp"
+#include "core/data_type/data_type_nullable.h"
+#include "core/typeid_cast.h"
+#include "core/value/variant/variant_parquet_encoding.h"
+#include "exec/common/format_ip.h"
+#include "exprs/function/parse/variant_jsonb_parse.h"
+#include "storage/segment/variant/v2/variant_assembler_internal.h"
+#include "util/utf8_check.h"
+
+namespace doris::segment_v2::variant_v2::variant_assembler_internal {
+namespace {
+
+class CellCursor {
+public:
+ explicit CellCursor(StringRef cell)
+ : _current(reinterpret_cast<const uint8_t*>(cell.data)),
_remaining(cell.size) {
+ if (cell.data == nullptr && cell.size != 0) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage cell has a null pointer for {}
bytes", cell.size);
+ }
+ }
+
+ template <typename T>
+ T read(std::string_view description) {
+ require(sizeof(T), description);
+ T value;
+ std::memcpy(&value, _current, sizeof(T));
+ _current += sizeof(T);
+ _remaining -= sizeof(T);
+ return value;
+ }
+
+ StringRef read_bytes(size_t size, std::string_view description) {
+ require(size, description);
+ const StringRef result {reinterpret_cast<const char*>(_current), size};
+ _current += size;
+ _remaining -= size;
+ return result;
+ }
+
+ size_t remaining() const noexcept { return _remaining; }
+ bool empty() const noexcept { return _remaining == 0; }
+
+private:
+ void require(size_t size, std::string_view description) const {
+ if (size > _remaining) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Truncated Variant storage cell while reading {}:
need {} bytes, "
+ "have {}",
+ description, size, _remaining);
+ }
+ }
+
+ const uint8_t* _current;
+ size_t _remaining;
+};
+
+template <typename Integer>
+uint32_t decimal_digits(Integer value) {
+ uint32_t result = 0;
+ do {
+ value /= 10;
+ ++result;
+ } while (value != 0);
+ return result;
+}
+
+template <typename Integer>
+void validate_decimal(Integer value, uint8_t precision, uint8_t scale, uint8_t
maximum,
+ std::string_view description) {
+ if (precision == 0 || precision > maximum || scale > precision) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage {} precision/scale {}/{} is invalid",
description,
+ precision, scale);
+ }
+ if (decimal_digits(value) > precision) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage {} value exceeds declared precision
{}", description,
+ precision);
+ }
+}
+
+void validate_depth(uint32_t depth) {
+ if (depth > VARIANT_MAX_NESTING_DEPTH) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage cell exceeds maximum nesting depth
{}",
+ VARIANT_MAX_NESTING_DEPTH);
+ }
+}
+
+void validate_container_depth(uint32_t depth) {
+ if (depth >= VARIANT_MAX_NESTING_DEPTH) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage container exceeds maximum nesting
depth {}",
+ VARIANT_MAX_NESTING_DEPTH);
+ }
+}
+
+template <typename DateValue>
+int32_t storage_date_days(DateValue value, std::string_view description) {
+ if (!value.is_valid_date()) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid {} in Variant storage
cell", description);
+ }
+ return variant_days_since_epoch(value, 0, description);
+}
+
+template <typename DateTimeValue>
+int64_t storage_timestamp_micros(DateTimeValue value, std::string_view
description) {
+ if (!value.is_valid_date()) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid {} in Variant storage
cell", description);
+ }
+ return variant_timestamp_micros(value, 0, description);
+}
+
+VecDateTimeValue read_legacy_temporal(CellCursor& cursor, FieldType type) {
+ const bool is_date = type == FieldType::OLAP_FIELD_TYPE_DATE;
+ DORIS_CHECK(is_date || type == FieldType::OLAP_FIELD_TYPE_DATETIME);
+ const auto value = cursor.read<VecDateTimeValue>(is_date ? "legacy DATE" :
"legacy DATETIME");
+ if (is_date) {
+ static_cast<void>(storage_date_days(value, "legacy DATE"));
+ } else {
+ static_cast<void>(storage_timestamp_micros(value, "legacy DATETIME"));
+ }
+ return value;
+}
+
+struct LegacyDecimalCell {
+ uint8_t precision;
+ uint8_t scale;
+ __int128 value;
+};
+
+LegacyDecimalCell read_legacy_decimal(CellCursor& cursor) {
+ LegacyDecimalCell result {
+ .precision = cursor.read<uint8_t>("legacy DecimalV2 precision"),
+ .scale = cursor.read<uint8_t>("legacy DecimalV2 scale"),
+ .value = cursor.read<__int128>("legacy DecimalV2 value"),
+ };
+ if (result.precision == 0 || result.precision > DecimalV2Value::PRECISION
||
+ result.scale > result.precision) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage legacy DecimalV2 precision/scale
{}/{} is invalid",
+ result.precision, result.scale);
+ }
+ // DecimalV2 always stores a fixed nine-digit fractional coefficient.
Preserve the serialized
+ // SerDe metadata for typed identity, but do not use it to rescale the
coefficient.
+ validate_decimal(result.value, DecimalV2Value::PRECISION,
DecimalV2Value::SCALE,
+ DecimalV2Value::PRECISION, "legacy DecimalV2");
+ return result;
+}
+
+void add_largeint(VariantBatchBuilder::Row& output, __int128 value) {
+ output.add_largeint(value);
+}
+
+void add_decimal256(VariantBatchBuilder::Row& output, wide::Int256 value,
uint8_t scale) {
+ const std::string text = Decimal256 {value}.to_string(scale);
+ output.add_string(StringRef(text));
+}
+
+void add_ipv4(VariantBatchBuilder::Row& output, IPv4 value) {
+ std::array<char, IPV4_MAX_TEXT_LENGTH + 1> buffer {};
+ char* end = buffer.data();
+ const auto* address = reinterpret_cast<const unsigned char*>(&value);
+ format_ipv4(address, end);
+ output.add_string({buffer.data(), static_cast<size_t>(end -
buffer.data())});
+}
+
+void add_ipv6(VariantBatchBuilder::Row& output, IPv6 value) {
+ std::array<char, IPV6_MAX_TEXT_LENGTH + 1> buffer {};
+ char* end = buffer.data();
+ format_ipv6(reinterpret_cast<unsigned char*>(&value), end);
+ output.add_string({buffer.data(), static_cast<size_t>(end -
buffer.data())});
+}
+
+StringRef read_storage_string(CellCursor& cursor) {
+ const size_t size = cursor.read<size_t>("string size");
+ const StringRef value = cursor.read_bytes(size, "string payload");
+ if (value.size != 0 && !validate_utf8(value.data, value.size)) {
+ throw Exception(ErrorCode::CORRUPTION, "Variant storage string is not
valid UTF-8");
+ }
+ return value;
+}
+
+// Keep the exhaustive storage FieldType decoder together so validation and
byte consumption for
+// every wire tag remain auditable in one dispatch table.
+// NOLINTNEXTLINE(readability-function-size)
+void decode_storage_value(CellCursor& cursor, VariantBatchBuilder::Row&
output, uint32_t depth) {
+ validate_depth(depth);
+ const auto type = static_cast<FieldType>(cursor.read<uint8_t>("field
type"));
+ switch (type) {
+ case FieldType::OLAP_FIELD_TYPE_NONE:
+ output.add_null();
+ return;
+ case FieldType::OLAP_FIELD_TYPE_BOOL: {
+ const uint8_t value = cursor.read<uint8_t>("boolean");
+ if (value > 1) {
+ throw Exception(ErrorCode::CORRUPTION, "Invalid Variant storage
boolean byte {}",
+ value);
+ }
+ output.add_bool(value != 0);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_TINYINT:
+ output.add_int(cursor.read<int8_t>("tinyint"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT:
+ output.add_int(cursor.read<int16_t>("smallint"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_INT:
+ output.add_int(cursor.read<int32_t>("int"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_BIGINT:
+ output.add_int(cursor.read<int64_t>("bigint"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_LARGEINT:
+ add_largeint(output, cursor.read<__int128>("largeint"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_FLOAT:
+ output.add_float(cursor.read<float>("float"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_DOUBLE:
+ output.add_double(cursor.read<double>("double"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_STRING:
+ output.add_string(read_storage_string(cursor));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_JSONB: {
+ const size_t size = cursor.read<size_t>("JSONB size");
+ jsonb_to_variant(cursor.read_bytes(size, "JSONB payload"), output,
depth);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_ARRAY: {
+ validate_container_depth(depth);
+ const size_t count = cursor.read<size_t>("array element count");
+ if (count > cursor.remaining()) {
+ throw Exception(ErrorCode::CORRUPTION,
+ "Variant storage array count {} exceeds remaining
{} bytes", count,
+ cursor.remaining());
+ }
+ auto array = output.start_array();
+ for (size_t index = 0; index < count; ++index) {
+ decode_storage_value(cursor, output, depth + 1);
+ }
+ array.finish();
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_IPV4:
+ add_ipv4(output, cursor.read<IPv4>("IPv4"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_IPV6:
+ add_ipv6(output, cursor.read<IPv6>("IPv6"));
+ return;
+ case FieldType::OLAP_FIELD_TYPE_DATE: {
+ const auto value = read_legacy_temporal(cursor, type);
+ output.add_date(storage_date_days(value, "legacy DATE"));
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIME: {
+ const auto value = read_legacy_temporal(cursor, type);
+ output.add_timestamp_micros(storage_timestamp_micros(value, "legacy
DATETIME"), false);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATEV2: {
+ const UInt32 raw = cursor.read<UInt32>("DateV2");
+ const auto value = binary_cast<UInt32,
DateV2Value<DateV2ValueType>>(raw);
+ output.add_date(storage_date_days(value, "DATEV2"));
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DATETIMEV2:
+ case FieldType::OLAP_FIELD_TYPE_TIMESTAMPTZ: {
+ const uint8_t scale = cursor.read<uint8_t>("timestamp scale");
+ if (scale > 6) {
+ throw Exception(ErrorCode::CORRUPTION, "Variant storage timestamp
scale {} exceeds 6",
+ scale);
+ }
+ const UInt64 raw = cursor.read<UInt64>("timestamp value");
+ if (type == FieldType::OLAP_FIELD_TYPE_DATETIMEV2) {
+ const auto value = binary_cast<UInt64,
DateV2Value<DateTimeV2ValueType>>(raw);
+ output.add_timestamp_micros(storage_timestamp_micros(value,
"DATETIMEV2"), false);
+ } else {
+ const auto value = binary_cast<UInt64, TimestampTzValue>(raw);
+ output.add_timestamp_micros(storage_timestamp_micros(value,
"TIMESTAMPTZ"), true);
+ }
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL: {
+ const auto value = read_legacy_decimal(cursor);
+ output.add_decimal(value.value, DecimalV2Value::SCALE, 16);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL32: {
+ const uint8_t precision = cursor.read<uint8_t>("Decimal32 precision");
+ const uint8_t scale = cursor.read<uint8_t>("Decimal32 scale");
+ const int32_t value = cursor.read<int32_t>("Decimal32 value");
+ validate_decimal(value, precision, scale, 9, "Decimal32");
+ output.add_decimal(value, scale, 4);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL64: {
+ const uint8_t precision = cursor.read<uint8_t>("Decimal64 precision");
+ const uint8_t scale = cursor.read<uint8_t>("Decimal64 scale");
+ const int64_t value = cursor.read<int64_t>("Decimal64 value");
+ validate_decimal(value, precision, scale, 18, "Decimal64");
+ output.add_decimal(value, scale, 8);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL128I: {
+ const uint8_t precision = cursor.read<uint8_t>("Decimal128 precision");
+ const uint8_t scale = cursor.read<uint8_t>("Decimal128 scale");
+ const __int128 value = cursor.read<__int128>("Decimal128 value");
+ validate_decimal(value, precision, scale, 38, "Decimal128");
+ output.add_decimal(value, scale, 16);
+ return;
+ }
+ case FieldType::OLAP_FIELD_TYPE_DECIMAL256: {
+ const uint8_t precision = cursor.read<uint8_t>("Decimal256 precision");
+ const uint8_t scale = cursor.read<uint8_t>("Decimal256 scale");
+ const wide::Int256 value = cursor.read<wide::Int256>("Decimal256
value");
+ validate_decimal(value, precision, scale, 76, "Decimal256");
+ add_decimal256(output, value, scale);
+ return;
+ }
+ default:
+ throw Exception(ErrorCode::CORRUPTION, "Unknown Variant storage
FieldType {}",
+ static_cast<uint8_t>(type));
+ }
+}
+
+template <typename ColumnType>
+bool matches(const IColumn* column) {
+ return check_and_get_column<ColumnType>(column) != nullptr;
+}
+
+bool supported_column_shape(PrimitiveType type, const IColumn* column) {
+ switch (type) {
+ case TYPE_BOOLEAN:
+ return matches<ColumnUInt8>(column);
+ case TYPE_TINYINT:
+ return matches<ColumnInt8>(column);
+ case TYPE_SMALLINT:
+ return matches<ColumnInt16>(column);
+ case TYPE_INT:
+ return matches<ColumnInt32>(column);
+ case TYPE_BIGINT:
+ return matches<ColumnInt64>(column);
+ case TYPE_LARGEINT:
+ return matches<ColumnInt128>(column);
+ case TYPE_FLOAT:
+ return matches<ColumnFloat32>(column);
+ case TYPE_DOUBLE:
+ return matches<ColumnFloat64>(column);
+ case TYPE_DECIMALV2:
+ return matches<ColumnDecimal128V2>(column);
+ case TYPE_DECIMAL32:
+ return matches<ColumnDecimal32>(column);
+ case TYPE_DECIMAL64:
+ return matches<ColumnDecimal64>(column);
+ case TYPE_DECIMAL128I:
+ return matches<ColumnDecimal128V3>(column);
+ case TYPE_DECIMAL256:
+ return matches<ColumnDecimal256>(column);
+ case TYPE_DATE:
+ return matches<ColumnDate>(column);
+ case TYPE_DATEV2:
+ return matches<ColumnDateV2>(column);
+ case TYPE_DATETIME:
+ return matches<ColumnDateTime>(column);
+ case TYPE_DATETIMEV2:
+ return matches<ColumnDateTimeV2>(column);
+ case TYPE_TIMESTAMPTZ:
+ return matches<ColumnTimeStampTz>(column);
+ case TYPE_CHAR:
+ case TYPE_VARCHAR:
+ case TYPE_STRING:
+ case TYPE_JSONB:
+ return matches<ColumnString>(column);
+ case TYPE_IPV4:
+ return matches<ColumnIPv4>(column);
+ case TYPE_IPV6:
+ return matches<ColumnIPv6>(column);
+ case TYPE_ARRAY:
+ return matches<ColumnArray>(column);
+ default:
+ return false;
+ }
+}
+
+// This is a flat PrimitiveType dispatch; splitting it would scatter the
concrete-column contract.
+// NOLINTNEXTLINE(readability-function-size)
+void append_materialized_scalar(const PreparedColumn& column, size_t row,
Review Comment:
[P2] 请复用已有的 V2 typed scalar dispatch。
`core/column/variant_v2/column_variant_v2_typed_column.h` 已集中定义
`is_supported_variant_typed_identity`、`dispatch_variant_typed_column` 和
`with_variant_typed_scalar`，`cast_array_to_variant.cpp` 也在复用它。这里的
`supported_column_shape` 加这个 switch 又声明了一遍 PrimitiveType -> concrete column ->
Variant scalar 映射。
建议共享 scalar 集合直接走现有 dispatch，只隔离 assembler 特有的 JSONB/array/Decimal256 或
error policy；否则新增类型时 cast 与 segment read 很容易出现两套行为。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
