be/src/format/parquet/parquet_column_convert.cpp

Source
// 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 "format/parquet/parquet_column_convert.h"

#include <cctz/time_zone.h>
#include <glog/logging.h>

#include <utility>

#include "common/cast_set.h"
#include "core/column/column_fixed_length_object.h"
#include "core/column/column_nullable.h"
#include "core/data_type/data_type_fixed_length_object.h"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/primitive_type.h"
#include "core/value/uuid_value.h"
#include "exprs/function/geo/functions_geo.h"

namespace doris::parquet {
const cctz::time_zone ConvertParams::utc0 = cctz::utc_time_zone();

#define FOR_LOGICAL_DECIMAL_TYPES(M) \
    M(TYPE_DECIMAL32)                \
    M(TYPE_DECIMAL64)                \
    M(TYPE_DECIMAL128I)              \
    M(TYPE_DECIMAL256)

bool PhysicalToLogicalConverter::is_parquet_native_type(PrimitiveType type) {
    switch (type) {
    case TYPE_BOOLEAN:
    case TYPE_INT:
    case TYPE_BIGINT:
    case TYPE_FLOAT:
    case TYPE_DOUBLE:
    case TYPE_STRING:
    case TYPE_CHAR:
    case TYPE_VARCHAR:
        return true;
    default:
        return false;
    }
}

bool PhysicalToLogicalConverter::is_decimal_type(doris::PrimitiveType type) {
    switch (type) {
    case TYPE_DECIMAL32:
    case TYPE_DECIMAL64:
    case TYPE_DECIMAL128I:
    case TYPE_DECIMAL256:
    case TYPE_DECIMALV2:
        return true;
    default:
        return false;
    }
}

ColumnPtr PhysicalToLogicalConverter::get_physical_column(tparquet::Type::type src_physical_type,
                                                          DataTypePtr src_logical_type,
                                                          ColumnPtr& dst_logical_column,
                                                          const DataTypePtr& dst_logical_type,
                                                          bool is_dict_filter) {
    if (is_dict_filter) {
        src_physical_type = tparquet::Type::INT32;
        src_logical_type = DataTypeFactory::instance().create_data_type(
                PrimitiveType::TYPE_INT, dst_logical_type->is_nullable());
    }

    if (!_convert_params->is_type_compatibility && is_consistent() &&
        _logical_converter->is_consistent()) {
        if (_cached_src_physical_type == nullptr) {
            _cached_src_physical_type = dst_logical_type->is_nullable()
                                                ? make_nullable(src_logical_type)
                                                : remove_nullable(src_logical_type);
        }
        return dst_logical_column;
    }

    if (!_cached_src_physical_column) {
        switch (src_physical_type) {
        case tparquet::Type::type::BOOLEAN:
            _cached_src_physical_type = std::make_shared<DataTypeUInt8>();
            break;
        case tparquet::Type::type::INT32:
            _cached_src_physical_type = std::make_shared<DataTypeInt32>();
            break;
        case tparquet::Type::type::INT64:
            _cached_src_physical_type = std::make_shared<DataTypeInt64>();
            break;
        case tparquet::Type::type::FLOAT:
            _cached_src_physical_type = std::make_shared<DataTypeFloat32>();
            break;
        case tparquet::Type::type::DOUBLE:
            _cached_src_physical_type = std::make_shared<DataTypeFloat64>();
            break;
        case tparquet::Type::type::BYTE_ARRAY:
            _cached_src_physical_type = std::make_shared<DataTypeString>();
            break;
        case tparquet::Type::type::FIXED_LEN_BYTE_ARRAY:
            _cached_src_physical_type = std::make_shared<DataTypeFixedLengthObject>();
            break;
        case tparquet::Type::type::INT96:
            _cached_src_physical_type = std::make_shared<DataTypeInt8>();
            break;
        }
        const bool is_fixed_length_byte_array =
                src_physical_type == tparquet::Type::type::FIXED_LEN_BYTE_ARRAY;
        if (dst_logical_type->is_nullable()) {
            MutableColumnPtr nested_physical_column;
            if (is_fixed_length_byte_array) {
                nested_physical_column = ColumnFixedLengthObject::create(
                        _convert_params->field_schema->parquet_schema.type_length);
            } else {
                nested_physical_column = _cached_src_physical_type->create_column();
            }
            _cached_src_physical_column = ColumnNullable::create(std::move(nested_physical_column),
                                                                 ColumnUInt8::create());
            _cached_src_physical_type = make_nullable(_cached_src_physical_type);
        } else {
            if (is_fixed_length_byte_array) {
                _cached_src_physical_column = ColumnFixedLengthObject::create(
                        _convert_params->field_schema->parquet_schema.type_length);
            } else {
                _cached_src_physical_column = _cached_src_physical_type->create_column();
            }
        }
    }
    // remove the old cached data
    auto cached_src_physical_column = IColumn::mutate(std::move(_cached_src_physical_column));
    cached_src_physical_column->clear();
    _cached_src_physical_column = std::move(cached_src_physical_column);

    return _cached_src_physical_column;
}

static void get_decimal_converter(const FieldSchema* field_schema, DataTypePtr src_logical_type,
                                  const DataTypePtr& dst_logical_type,
                                  ConvertParams* convert_params,
                                  std::unique_ptr<PhysicalToLogicalConverter>& physical_converter) {
    const tparquet::SchemaElement& parquet_schema = field_schema->parquet_schema;
    if (is_decimal(dst_logical_type->get_primitive_type())) {
        src_logical_type = create_decimal(parquet_schema.precision, parquet_schema.scale, false);
    }

    tparquet::Type::type src_physical_type = parquet_schema.type;
    PrimitiveType src_logical_primitive = src_logical_type->get_primitive_type();

    if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
        switch (src_logical_primitive) {
#define DISPATCH(LOGICAL_PTYPE)                                                     \
    case LOGICAL_PTYPE: {                                                           \
        physical_converter.reset(                                                   \
                new FixedSizeToDecimal<LOGICAL_PTYPE>(parquet_schema.type_length)); \
        break;                                                                      \
    }
            FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
#undef DISPATCH
        default:
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else if (src_physical_type == tparquet::Type::BYTE_ARRAY) {
        switch (src_logical_primitive) {
#define DISPATCH(LOGICAL_PTYPE)                                         \
    case LOGICAL_PTYPE: {                                               \
        physical_converter.reset(new StringToDecimal<LOGICAL_PTYPE>()); \
        break;                                                          \
    }
            FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
#undef DISPATCH
        default:
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else if (src_physical_type == tparquet::Type::INT32 ||
               src_physical_type == tparquet::Type::INT64) {
        switch (src_logical_primitive) {
#define DISPATCH(LOGICAL_PTYPE)                                                          \
    case LOGICAL_PTYPE: {                                                                \
        if (src_physical_type == tparquet::Type::INT32) {                                \
            physical_converter.reset(new NumberToDecimal<TYPE_INT, LOGICAL_PTYPE>());    \
        } else {                                                                         \
            physical_converter.reset(new NumberToDecimal<TYPE_BIGINT, LOGICAL_PTYPE>()); \
        }                                                                                \
        break;                                                                           \
    }
            FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
#undef DISPATCH
        default:
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else {
        physical_converter =
                std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
    }
}

namespace {

// V1 decodes an annotated spatial BYTE_ARRAY into ColumnString first. Materialize
// its raw WKB into ColumnSpatial here; unannotated Iceberg binary fields instead
// use BinaryToSpatialConverter after the consistent physical conversion.
class SpatialPhysicalConverter final : public PhysicalToLogicalConverter {
public:
    explicit SpatialPhysicalConverter(DataTypePtr spatial_type)
            : _spatial_type(std::move(spatial_type)) {}

    Status physical_convert(ColumnPtr& source, ColumnPtr& destination) override {
        return append_spatial_wkb(source, *get_mutable_inner_column(destination), _spatial_type);
    }

private:
    DataTypePtr _spatial_type;
};

class NativeUUIDConverter final : public PhysicalToLogicalConverter {
public:
    Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override {
        const auto from_col = remove_nullable(src_physical_col);
        const auto src_data = get_fixed_length_physical_data(*from_col, UUIDValue::BINARY_LENGTH);
        auto& values =
                assert_cast<ColumnUUID&>(*get_mutable_inner_column(src_logical_column)).get_data();
        for (size_t i = 0; i < src_data.rows; ++i) {
            // File UUIDs use network byte order, unlike the native column's integer layout.
            values.push_back(UUIDValue::from_big_endian(reinterpret_cast<const uint8_t*>(
                    src_data.data + i * UUIDValue::BINARY_LENGTH)));
        }
        return Status::OK();
    }
};

class UUIDStringConverter final : public PhysicalToLogicalConverter {
public:
    Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override {
        const auto from_col = remove_nullable(src_physical_col);
        const auto src_data = get_fixed_length_physical_data(*from_col, 16);
        auto& strings = assert_cast<ColumnString&>(*get_mutable_inner_column(src_logical_column));
        for (size_t i = 0; i < src_data.rows; ++i) {
            const auto value = UUIDValue::from_big_endian(
                    reinterpret_cast<const uint8_t*>(src_data.data + i * 16));
            const auto text = UUIDValue::to_string(value);
            strings.insert_data(text.data(), text.size());
        }
        return Status::OK();
    }
};

} // namespace

std::unique_ptr<PhysicalToLogicalConverter> PhysicalToLogicalConverter::get_converter(
        const FieldSchema* field_schema, DataTypePtr src_logical_type,
        const DataTypePtr& dst_logical_type, const cctz::time_zone* ctz, bool is_dict_filter,
        bool preserve_binary_uuid) {
    std::unique_ptr<ConvertParams> convert_params = std::make_unique<ConvertParams>();
    const tparquet::SchemaElement& parquet_schema = field_schema->parquet_schema;
    convert_params->init(field_schema, ctz);
    tparquet::Type::type src_physical_type = parquet_schema.type;
    std::unique_ptr<PhysicalToLogicalConverter> physical_converter;
    if (is_dict_filter) {
        src_physical_type = tparquet::Type::INT32;
        src_logical_type = DataTypeFactory::instance().create_data_type(
                PrimitiveType::TYPE_INT, dst_logical_type->is_nullable());
    }
    // Hive BINARY columns retain annotated WKB as opaque bytes. Only this explicit binary
    // target uses the physical STRING carrier; native spatial targets keep annotation metadata.
    const auto source_primitive = src_logical_type->get_primitive_type();
    if (!is_dict_filter && src_physical_type == tparquet::Type::BYTE_ARRAY &&
        (source_primitive == TYPE_GEOMETRY || source_primitive == TYPE_GEOGRAPHY) &&
        dst_logical_type->get_primitive_type() == TYPE_VARBINARY) {
        src_logical_type = DataTypeFactory::instance().create_data_type(
                TYPE_STRING, src_logical_type->is_nullable());
    }
    PrimitiveType src_logical_primitive = src_logical_type->get_primitive_type();

    if (field_schema->is_type_compatibility) {
        if (src_logical_primitive == TYPE_SMALLINT) {
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_SMALLINT>>();
        } else if (src_logical_primitive == TYPE_INT) {
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_INT>>();
        } else if (src_logical_primitive == TYPE_BIGINT) {
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_BIGINT>>();
        } else if (src_logical_primitive == TYPE_LARGEINT) {
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_LARGEINT>>();
        } else {
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else if (is_parquet_native_type(src_logical_primitive)) {
        bool is_string_logical_type = is_string_type(src_logical_primitive);
        if (is_string_logical_type && src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
            if (parquet_schema.logicalType.__isset.UUID && !preserve_binary_uuid) {
                // Ordinary Parquet scans expose canonical text, as in File Scanner V2.
                // Iceberg retains its raw 16-byte STRING carrier for equality deletes/defaults.
                if (parquet_schema.type_length != UUIDValue::BINARY_LENGTH) {
                    physical_converter = std::make_unique<UnsupportedConverter>(src_physical_type,
                                                                                src_logical_type);
                } else {
                    physical_converter = std::make_unique<UUIDStringConverter>();
                }
            } else {
                physical_converter =
                        std::make_unique<FixedSizeBinaryConverter>(parquet_schema.type_length);
            }
        } else if (src_logical_primitive == TYPE_FLOAT &&
                   src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY &&
                   parquet_schema.logicalType.__isset.FLOAT16) {
            physical_converter =
                    std::make_unique<Float16PhysicalConverter>(parquet_schema.type_length);
        } else {
            physical_converter = std::make_unique<ConsistentPhysicalConverter>();
        }
    } else if (src_logical_primitive == TYPE_TINYINT) {
        physical_converter = std::make_unique<LittleIntPhysicalConverter<TYPE_TINYINT>>();
    } else if (src_logical_primitive == TYPE_SMALLINT) {
        physical_converter = std::make_unique<LittleIntPhysicalConverter<TYPE_SMALLINT>>();
    } else if (is_decimal_type(src_logical_primitive)) {
        get_decimal_converter(field_schema, src_logical_type, dst_logical_type,
                              convert_params.get(), physical_converter);
    } else if (src_logical_primitive == TYPE_DATEV2) {
        physical_converter = std::make_unique<Int32ToDate>();
    } else if (src_logical_primitive == TYPE_DATETIMEV2) {
        if (src_physical_type == tparquet::Type::INT96) {
            // int96 only stores nanoseconds in standard parquet file
            convert_params->reset_time_scale_if_missing(9);
            physical_converter = std::make_unique<Int96toTimestamp>();
        } else if (src_physical_type == tparquet::Type::INT64) {
            convert_params->reset_time_scale_if_missing(src_logical_type->get_scale());
            physical_converter = std::make_unique<Int64ToTimestamp>();
        } else {
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else if (src_logical_primitive == TYPE_UUID) {
        if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY &&
            std::cmp_equal(parquet_schema.type_length, UUIDValue::BINARY_LENGTH)) {
            physical_converter = std::make_unique<NativeUUIDConverter>();
        } else {
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else if (src_logical_primitive == TYPE_GEOMETRY || src_logical_primitive == TYPE_GEOGRAPHY) {
        if (src_physical_type == tparquet::Type::BYTE_ARRAY) {
            physical_converter = std::make_unique<SpatialPhysicalConverter>(src_logical_type);
        } else {
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else if (src_logical_primitive == TYPE_VARBINARY) {
        if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
            DCHECK(parquet_schema.logicalType.__isset.UUID) << parquet_schema.name;
            physical_converter =
                    std::make_unique<UUIDVarBinaryConverter>(parquet_schema.type_length);
        } else {
            DCHECK(src_physical_type == tparquet::Type::BYTE_ARRAY) << src_physical_type;
            physical_converter = std::make_unique<ConsistentPhysicalConverter>();
        }
    } else if (src_logical_primitive == TYPE_TIMESTAMPTZ) {
        if (src_physical_type == tparquet::Type::INT96) {
            physical_converter = std::make_unique<Int96toTimestampTz>();
        } else if (src_physical_type == tparquet::Type::INT64) {
            DCHECK(src_physical_type == tparquet::Type::INT64) << src_physical_type;
            DCHECK(parquet_schema.logicalType.__isset.TIMESTAMP) << parquet_schema.name;
            physical_converter = std::make_unique<Int64ToTimestampTz>();
        } else {
            physical_converter =
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
        }
    } else {
        physical_converter =
                std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
    }

    if (physical_converter->support()) {
        physical_converter->_convert_params = std::move(convert_params);
        physical_converter->_logical_converter = converter::ColumnTypeConverter::get_converter(
                src_logical_type, dst_logical_type, converter::FileFormat::PARQUET);
        if (!physical_converter->_logical_converter->support()) {
            physical_converter = std::make_unique<UnsupportedConverter>(
                    "Unsupported type change: " +
                    physical_converter->_logical_converter->get_error_msg());
        }
    }
    return physical_converter;
}

} // namespace doris::parquet

Coverage Report

Created: 2026-10-27 23:51

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "format/parquet/parquet_column_convert.h"
19
20		#include <cctz/time_zone.h>
21		#include <glog/logging.h>
22
23		#include <utility>
24
25		#include "common/cast_set.h"
26		#include "core/column/column_fixed_length_object.h"
27		#include "core/column/column_nullable.h"
28		#include "core/data_type/data_type_fixed_length_object.h"
29		#include "core/data_type/data_type_nullable.h"
30		#include "core/data_type/define_primitive_type.h"
31		#include "core/data_type/primitive_type.h"
32		#include "core/value/uuid_value.h"
33		#include "exprs/function/geo/functions_geo.h"
34
35		namespace doris::parquet {
36		const cctz::time_zone ConvertParams::utc0 = cctz::utc_time_zone();
37
38		#define FOR_LOGICAL_DECIMAL_TYPES(M) \
39	0	M(TYPE_DECIMAL32) \
40	4	M(TYPE_DECIMAL64) \
41	4	M(TYPE_DECIMAL128I) \
42	0	M(TYPE_DECIMAL256)
43
44	402	bool PhysicalToLogicalConverter::is_parquet_native_type(PrimitiveType type) {
45	402	switch (type) {
46	12	case TYPE_BOOLEAN:
47	195	case TYPE_INT:
48	233	case TYPE_BIGINT:
49	263	case TYPE_FLOAT:
50	282	case TYPE_DOUBLE:
51	344	case TYPE_STRING:
52	344	case TYPE_CHAR:
53	344	case TYPE_VARCHAR:
54	344	return true;
55	58	default:
56	58	return false;
57	402	}
58	402	}
59
60	41	bool PhysicalToLogicalConverter::is_decimal_type(doris::PrimitiveType type) {
61	41	switch (type) {
62	0	case TYPE_DECIMAL32:
63	4	case TYPE_DECIMAL64:
64	4	case TYPE_DECIMAL128I:
65	4	case TYPE_DECIMAL256:
66	4	case TYPE_DECIMALV2:
67	4	return true;
68	37	default:
69	37	return false;
70	41	}
71	41	}
72
73		ColumnPtr PhysicalToLogicalConverter::get_physical_column(tparquet::Type::type src_physical_type,
74		DataTypePtr src_logical_type,
75		ColumnPtr& dst_logical_column,
76		const DataTypePtr& dst_logical_type,
77	422	bool is_dict_filter) {
78	422	if (is_dict_filter) {
79	1	src_physical_type = tparquet::Type::INT32;
80	1	src_logical_type = DataTypeFactory::instance().create_data_type(
81	1	PrimitiveType::TYPE_INT, dst_logical_type->is_nullable());
82	1	}
83
84	422	if (!_convert_params->is_type_compatibility && is_consistent() &&
85	422	_logical_converter->is_consistent()) {
86	370	if (_cached_src_physical_type == nullptr) {
87	226	_cached_src_physical_type = dst_logical_type->is_nullable()
88	226	? make_nullable(src_logical_type)
89	226	: remove_nullable(src_logical_type);
90	226	}
91	370	return dst_logical_column;
92	370	}
93
94	52	if (!_cached_src_physical_column) {
95	41	switch (src_physical_type) {
96	0	case tparquet::Type::type::BOOLEAN:
97	0	_cached_src_physical_type = std::make_shared<DataTypeUInt8>();
98	0	break;
99	23	case tparquet::Type::type::INT32:
100	23	_cached_src_physical_type = std::make_shared<DataTypeInt32>();
101	23	break;
102	6	case tparquet::Type::type::INT64:
103	6	_cached_src_physical_type = std::make_shared<DataTypeInt64>();
104	6	break;
105	0	case tparquet::Type::type::FLOAT:
106	0	_cached_src_physical_type = std::make_shared<DataTypeFloat32>();
107	0	break;
108	0	case tparquet::Type::type::DOUBLE:
109	0	_cached_src_physical_type = std::make_shared<DataTypeFloat64>();
110	0	break;
111	8	case tparquet::Type::type::BYTE_ARRAY:
112	8	_cached_src_physical_type = std::make_shared<DataTypeString>();
113	8	break;
114	4	case tparquet::Type::type::FIXED_LEN_BYTE_ARRAY:
115	4	_cached_src_physical_type = std::make_shared<DataTypeFixedLengthObject>();
116	4	break;
117	0	case tparquet::Type::type::INT96:
118	0	_cached_src_physical_type = std::make_shared<DataTypeInt8>();
119	0	break;
120	41	}
121	41	const bool is_fixed_length_byte_array =
122	41	src_physical_type == tparquet::Type::type::FIXED_LEN_BYTE_ARRAY;
123	41	if (dst_logical_type->is_nullable()) {
124	39	MutableColumnPtr nested_physical_column;
125	39	if (is_fixed_length_byte_array) {
126	4	nested_physical_column = ColumnFixedLengthObject::create(
127	4	_convert_params->field_schema->parquet_schema.type_length);
128	35	} else {
129	35	nested_physical_column = _cached_src_physical_type->create_column();
130	35	}
131	39	_cached_src_physical_column = ColumnNullable::create(std::move(nested_physical_column),
132	39	ColumnUInt8::create());
133	39	_cached_src_physical_type = make_nullable(_cached_src_physical_type);
134	39	} else {
135	2	if (is_fixed_length_byte_array) {
136	0	_cached_src_physical_column = ColumnFixedLengthObject::create(
137	0	_convert_params->field_schema->parquet_schema.type_length);
138	2	} else {
139	2	_cached_src_physical_column = _cached_src_physical_type->create_column();
140	2	}
141	2	}
142	41	}
143		// remove the old cached data
144	52	auto cached_src_physical_column = IColumn::mutate(std::move(_cached_src_physical_column));
145	52	cached_src_physical_column->clear();
146	52	_cached_src_physical_column = std::move(cached_src_physical_column);
147
148	52	return _cached_src_physical_column;
149	52	}
150
151		static void get_decimal_converter(const FieldSchema* field_schema, DataTypePtr src_logical_type,
152		const DataTypePtr& dst_logical_type,
153		ConvertParams* convert_params,
154	4	std::unique_ptr<PhysicalToLogicalConverter>& physical_converter) {
155	4	const tparquet::SchemaElement& parquet_schema = field_schema->parquet_schema;
156	4	if (is_decimal(dst_logical_type->get_primitive_type())) {
157	4	src_logical_type = create_decimal(parquet_schema.precision, parquet_schema.scale, false);
158	4	}
159
160	4	tparquet::Type::type src_physical_type = parquet_schema.type;
161	4	PrimitiveType src_logical_primitive = src_logical_type->get_primitive_type();
162
163	4	if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
164	2	switch (src_logical_primitive) {
165	0	#define DISPATCH(LOGICAL_PTYPE) \
166	2	case LOGICAL_PTYPE: { \
167	2	physical_converter.reset( \
168	2	new FixedSizeToDecimal<LOGICAL_PTYPE>(parquet_schema.type_length)); \
169	2	break; \
170	2	}
171	2	FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
172	0	#undef DISPATCH
173	0	default:
174	0	physical_converter =
175	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
176	2	}
177	2	} else if (src_physical_type == tparquet::Type::BYTE_ARRAY) {
178	0	switch (src_logical_primitive) {
179	0	#define DISPATCH(LOGICAL_PTYPE) \
180	0	case LOGICAL_PTYPE: { \
181	0	physical_converter.reset(new StringToDecimal<LOGICAL_PTYPE>()); \
182	0	break; \
183	0	}
184	0	FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
185	0	#undef DISPATCH
186	0	default:
187	0	physical_converter =
188	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
189	0	}
190	2	} else if (src_physical_type == tparquet::Type::INT32 \|\|
191	2	src_physical_type == tparquet::Type::INT64) {
192	2	switch (src_logical_primitive) {
193	0	#define DISPATCH(LOGICAL_PTYPE) \
194	2	case LOGICAL_PTYPE: { \
195	2	if (src_physical_type == tparquet::Type::INT32) { \
196	0	physical_converter.reset(new NumberToDecimal<TYPE_INT, LOGICAL_PTYPE>()); \
197	2	} else { \
198	2	physical_converter.reset(new NumberToDecimal<TYPE_BIGINT, LOGICAL_PTYPE>()); \
199	2	} \
200	2	break; \
201	2	}
202	2	FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
203	0	#undef DISPATCH
204	0	default:
205	0	physical_converter =
206	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
207	2	}
208	2	} else {
209	0	physical_converter =
210	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
211	0	}
212	4	}
213
214		namespace {
215
216		// V1 decodes an annotated spatial BYTE_ARRAY into ColumnString first. Materialize
217		// its raw WKB into ColumnSpatial here; unannotated Iceberg binary fields instead
218		// use BinaryToSpatialConverter after the consistent physical conversion.
219		class SpatialPhysicalConverter final : public PhysicalToLogicalConverter {
220		public:
221		explicit SpatialPhysicalConverter(DataTypePtr spatial_type)
222	2	: _spatial_type(std::move(spatial_type)) {}
223
224	2	Status physical_convert(ColumnPtr& source, ColumnPtr& destination) override {
225	2	return append_spatial_wkb(source, *get_mutable_inner_column(destination), _spatial_type);
226	2	}
227
228		private:
229		DataTypePtr _spatial_type;
230		};
231
232		class NativeUUIDConverter final : public PhysicalToLogicalConverter {
233		public:
234	5	Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override {
235	5	const auto from_col = remove_nullable(src_physical_col);
236	5	const auto src_data = get_fixed_length_physical_data(*from_col, UUIDValue::BINARY_LENGTH);
237	5	auto& values =
238	5	assert_cast<ColumnUUID&>(*get_mutable_inner_column(src_logical_column)).get_data();
239	19	for (size_t i = 0; i < src_data.rows; ++i) {
240		// File UUIDs use network byte order, unlike the native column's integer layout.
241	14	values.push_back(UUIDValue::from_big_endian(reinterpret_cast<const uint8_t*>(
242	14	src_data.data + i * UUIDValue::BINARY_LENGTH)));
243	14	}
244	5	return Status::OK();
245	5	}
246		};
247
248		class UUIDStringConverter final : public PhysicalToLogicalConverter {
249		public:
250	2	Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override {
251	2	const auto from_col = remove_nullable(src_physical_col);
252	2	const auto src_data = get_fixed_length_physical_data(*from_col, 16);
253	2	auto& strings = assert_cast<ColumnString&>(*get_mutable_inner_column(src_logical_column));
254	8	for (size_t i = 0; i < src_data.rows; ++i) {
255	6	const auto value = UUIDValue::from_big_endian(
256	6	reinterpret_cast<const uint8_t>(src_data.data + i 16));
257	6	const auto text = UUIDValue::to_string(value);
258	6	strings.insert_data(text.data(), text.size());
259	6	}
260	2	return Status::OK();
261	2	}
262		};
263
264		} // namespace
265
266		std::unique_ptr<PhysicalToLogicalConverter> PhysicalToLogicalConverter::get_converter(
267		const FieldSchema* field_schema, DataTypePtr src_logical_type,
268		const DataTypePtr& dst_logical_type, const cctz::time_zone* ctz, bool is_dict_filter,
269	402	bool preserve_binary_uuid) {
270	402	std::unique_ptr<ConvertParams> convert_params = std::make_unique<ConvertParams>();
271	402	const tparquet::SchemaElement& parquet_schema = field_schema->parquet_schema;
272	402	convert_params->init(field_schema, ctz);
273	402	tparquet::Type::type src_physical_type = parquet_schema.type;
274	402	std::unique_ptr<PhysicalToLogicalConverter> physical_converter;
275	402	if (is_dict_filter) {
276	1	src_physical_type = tparquet::Type::INT32;
277	1	src_logical_type = DataTypeFactory::instance().create_data_type(
278	1	PrimitiveType::TYPE_INT, dst_logical_type->is_nullable());
279	1	}
280		// Hive BINARY columns retain annotated WKB as opaque bytes. Only this explicit binary
281		// target uses the physical STRING carrier; native spatial targets keep annotation metadata.
282	402	const auto source_primitive = src_logical_type->get_primitive_type();
283	402	if (!is_dict_filter && src_physical_type == tparquet::Type::BYTE_ARRAY &&
284	402	(source_primitive == TYPE_GEOMETRY \|\| source_primitive == TYPE_GEOGRAPHY) &&
285	402	dst_logical_type->get_primitive_type() == TYPE_VARBINARY) {
286	4	src_logical_type = DataTypeFactory::instance().create_data_type(
287	4	TYPE_STRING, src_logical_type->is_nullable());
288	4	}
289	402	PrimitiveType src_logical_primitive = src_logical_type->get_primitive_type();
290
291	402	if (field_schema->is_type_compatibility) {
292	0	if (src_logical_primitive == TYPE_SMALLINT) {
293	0	physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_SMALLINT>>();
294	0	} else if (src_logical_primitive == TYPE_INT) {
295	0	physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_INT>>();
296	0	} else if (src_logical_primitive == TYPE_BIGINT) {
297	0	physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_BIGINT>>();
298	0	} else if (src_logical_primitive == TYPE_LARGEINT) {
299	0	physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_LARGEINT>>();
300	0	} else {
301	0	physical_converter =
302	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
303	0	}
304	402	} else if (is_parquet_native_type(src_logical_primitive)) {
305	344	bool is_string_logical_type = is_string_type(src_logical_primitive);
306	344	if (is_string_logical_type && src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
307	7	if (parquet_schema.logicalType.__isset.UUID && !preserve_binary_uuid) {
308		// Ordinary Parquet scans expose canonical text, as in File Scanner V2.
309		// Iceberg retains its raw 16-byte STRING carrier for equality deletes/defaults.
310	2	if (parquet_schema.type_length != UUIDValue::BINARY_LENGTH) {
311	1	physical_converter = std::make_unique<UnsupportedConverter>(src_physical_type,
312	1	src_logical_type);
313	1	} else {
314	1	physical_converter = std::make_unique<UUIDStringConverter>();
315	1	}
316	5	} else {
317	5	physical_converter =
318	5	std::make_unique<FixedSizeBinaryConverter>(parquet_schema.type_length);
319	5	}
320	337	} else if (src_logical_primitive == TYPE_FLOAT &&
321	337	src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY &&
322	337	parquet_schema.logicalType.__isset.FLOAT16) {
323	0	physical_converter =
324	0	std::make_unique<Float16PhysicalConverter>(parquet_schema.type_length);
325	337	} else {
326	337	physical_converter = std::make_unique<ConsistentPhysicalConverter>();
327	337	}
328	344	} else if (src_logical_primitive == TYPE_TINYINT) {
329	10	physical_converter = std::make_unique<LittleIntPhysicalConverter<TYPE_TINYINT>>();
330	48	} else if (src_logical_primitive == TYPE_SMALLINT) {
331	7	physical_converter = std::make_unique<LittleIntPhysicalConverter<TYPE_SMALLINT>>();
332	41	} else if (is_decimal_type(src_logical_primitive)) {
333	4	get_decimal_converter(field_schema, src_logical_type, dst_logical_type,
334	4	convert_params.get(), physical_converter);
335	37	} else if (src_logical_primitive == TYPE_DATEV2) {
336	7	physical_converter = std::make_unique<Int32ToDate>();
337	30	} else if (src_logical_primitive == TYPE_DATETIMEV2) {
338	16	if (src_physical_type == tparquet::Type::INT96) {
339		// int96 only stores nanoseconds in standard parquet file
340	1	convert_params->reset_time_scale_if_missing(9);
341	1	physical_converter = std::make_unique<Int96toTimestamp>();
342	15	} else if (src_physical_type == tparquet::Type::INT64) {
343	15	convert_params->reset_time_scale_if_missing(src_logical_type->get_scale());
344	15	physical_converter = std::make_unique<Int64ToTimestamp>();
345	15	} else {
346	0	physical_converter =
347	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
348	0	}
349	16	} else if (src_logical_primitive == TYPE_UUID) {
350	6	if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY &&
351	6	std::cmp_equal(parquet_schema.type_length, UUIDValue::BINARY_LENGTH)) {
352	5	physical_converter = std::make_unique<NativeUUIDConverter>();
353	5	} else {
354	1	physical_converter =
355	1	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
356	1	}
357	8	} else if (src_logical_primitive == TYPE_GEOMETRY \|\| src_logical_primitive == TYPE_GEOGRAPHY) {
358	2	if (src_physical_type == tparquet::Type::BYTE_ARRAY) {
359	2	physical_converter = std::make_unique<SpatialPhysicalConverter>(src_logical_type);
360	2	} else {
361	0	physical_converter =
362	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
363	0	}
364	6	} else if (src_logical_primitive == TYPE_VARBINARY) {
365	4	if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
366	0	DCHECK(parquet_schema.logicalType.__isset.UUID) << parquet_schema.name;
367	0	physical_converter =
368	0	std::make_unique<UUIDVarBinaryConverter>(parquet_schema.type_length);
369	4	} else {
370	4	DCHECK(src_physical_type == tparquet::Type::BYTE_ARRAY) << src_physical_type;
371	4	physical_converter = std::make_unique<ConsistentPhysicalConverter>();
372	4	}
373	4	} else if (src_logical_primitive == TYPE_TIMESTAMPTZ) {
374	2	if (src_physical_type == tparquet::Type::INT96) {
375	1	physical_converter = std::make_unique<Int96toTimestampTz>();
376	1	} else if (src_physical_type == tparquet::Type::INT64) {
377	1	DCHECK(src_physical_type == tparquet::Type::INT64) << src_physical_type;
378	1	DCHECK(parquet_schema.logicalType.__isset.TIMESTAMP) << parquet_schema.name;
379	1	physical_converter = std::make_unique<Int64ToTimestampTz>();
380	1	} else {
381	0	physical_converter =
382	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
383	0	}
384	2	} else {
385	0	physical_converter =
386	0	std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
387	0	}
388
389	402	if (physical_converter->support()) {
390	400	physical_converter->_convert_params = std::move(convert_params);
391	400	physical_converter->_logical_converter = converter::ColumnTypeConverter::get_converter(
392	400	src_logical_type, dst_logical_type, converter::FileFormat::PARQUET);
393	400	if (!physical_converter->_logical_converter->support()) {
394	0	physical_converter = std::make_unique<UnsupportedConverter>(
395	0	"Unsupported type change: " +
396	0	physical_converter->_logical_converter->get_error_msg());
397	0	}
398	400	}
399	402	return physical_converter;
400	402	}
401
402		} // namespace doris::parquet