Coverage Report

Created: 2026-10-28 16:17

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/parquet/parquet_column_convert.cpp
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
34
namespace doris::parquet {
35
const cctz::time_zone ConvertParams::utc0 = cctz::utc_time_zone();
36
37
#define FOR_LOGICAL_DECIMAL_TYPES(M) \
38
0
    M(TYPE_DECIMAL32)                \
39
0
    M(TYPE_DECIMAL64)                \
40
0
    M(TYPE_DECIMAL128I)              \
41
0
    M(TYPE_DECIMAL256)
42
43
78
bool PhysicalToLogicalConverter::is_parquet_native_type(PrimitiveType type) {
44
78
    switch (type) {
45
0
    case TYPE_BOOLEAN:
46
42
    case TYPE_INT:
47
43
    case TYPE_BIGINT:
48
54
    case TYPE_FLOAT:
49
54
    case TYPE_DOUBLE:
50
63
    case TYPE_STRING:
51
63
    case TYPE_CHAR:
52
63
    case TYPE_VARCHAR:
53
63
        return true;
54
15
    default:
55
15
        return false;
56
78
    }
57
78
}
58
59
15
bool PhysicalToLogicalConverter::is_decimal_type(doris::PrimitiveType type) {
60
15
    switch (type) {
61
0
    case TYPE_DECIMAL32:
62
0
    case TYPE_DECIMAL64:
63
0
    case TYPE_DECIMAL128I:
64
0
    case TYPE_DECIMAL256:
65
0
    case TYPE_DECIMALV2:
66
0
        return true;
67
15
    default:
68
15
        return false;
69
15
    }
70
15
}
71
72
ColumnPtr PhysicalToLogicalConverter::get_physical_column(tparquet::Type::type src_physical_type,
73
                                                          DataTypePtr src_logical_type,
74
                                                          ColumnPtr& dst_logical_column,
75
                                                          const DataTypePtr& dst_logical_type,
76
0
                                                          bool is_dict_filter) {
77
0
    if (is_dict_filter) {
78
0
        src_physical_type = tparquet::Type::INT32;
79
0
        src_logical_type = DataTypeFactory::instance().create_data_type(
80
0
                PrimitiveType::TYPE_INT, dst_logical_type->is_nullable());
81
0
    }
82
83
0
    if (!_convert_params->is_type_compatibility && is_consistent() &&
84
0
        _logical_converter->is_consistent()) {
85
0
        if (_cached_src_physical_type == nullptr) {
86
0
            _cached_src_physical_type = dst_logical_type->is_nullable()
87
0
                                                ? make_nullable(src_logical_type)
88
0
                                                : remove_nullable(src_logical_type);
89
0
        }
90
0
        return dst_logical_column;
91
0
    }
92
93
0
    if (!_cached_src_physical_column) {
94
0
        switch (src_physical_type) {
95
0
        case tparquet::Type::type::BOOLEAN:
96
0
            _cached_src_physical_type = std::make_shared<DataTypeUInt8>();
97
0
            break;
98
0
        case tparquet::Type::type::INT32:
99
0
            _cached_src_physical_type = std::make_shared<DataTypeInt32>();
100
0
            break;
101
0
        case tparquet::Type::type::INT64:
102
0
            _cached_src_physical_type = std::make_shared<DataTypeInt64>();
103
0
            break;
104
0
        case tparquet::Type::type::FLOAT:
105
0
            _cached_src_physical_type = std::make_shared<DataTypeFloat32>();
106
0
            break;
107
0
        case tparquet::Type::type::DOUBLE:
108
0
            _cached_src_physical_type = std::make_shared<DataTypeFloat64>();
109
0
            break;
110
0
        case tparquet::Type::type::BYTE_ARRAY:
111
0
            _cached_src_physical_type = std::make_shared<DataTypeString>();
112
0
            break;
113
0
        case tparquet::Type::type::FIXED_LEN_BYTE_ARRAY:
114
0
            _cached_src_physical_type = std::make_shared<DataTypeFixedLengthObject>();
115
0
            break;
116
0
        case tparquet::Type::type::INT96:
117
0
            _cached_src_physical_type = std::make_shared<DataTypeInt8>();
118
0
            break;
119
0
        }
120
0
        const bool is_fixed_length_byte_array =
121
0
                src_physical_type == tparquet::Type::type::FIXED_LEN_BYTE_ARRAY;
122
0
        if (dst_logical_type->is_nullable()) {
123
0
            MutableColumnPtr nested_physical_column;
124
0
            if (is_fixed_length_byte_array) {
125
0
                nested_physical_column = ColumnFixedLengthObject::create(
126
0
                        _convert_params->field_schema->parquet_schema.type_length);
127
0
            } else {
128
0
                nested_physical_column = _cached_src_physical_type->create_column();
129
0
            }
130
0
            _cached_src_physical_column = ColumnNullable::create(std::move(nested_physical_column),
131
0
                                                                 ColumnUInt8::create());
132
0
            _cached_src_physical_type = make_nullable(_cached_src_physical_type);
133
0
        } else {
134
0
            if (is_fixed_length_byte_array) {
135
0
                _cached_src_physical_column = ColumnFixedLengthObject::create(
136
0
                        _convert_params->field_schema->parquet_schema.type_length);
137
0
            } else {
138
0
                _cached_src_physical_column = _cached_src_physical_type->create_column();
139
0
            }
140
0
        }
141
0
    }
142
    // remove the old cached data
143
0
    auto cached_src_physical_column = IColumn::mutate(std::move(_cached_src_physical_column));
144
0
    cached_src_physical_column->clear();
145
0
    _cached_src_physical_column = std::move(cached_src_physical_column);
146
147
0
    return _cached_src_physical_column;
148
0
}
149
150
static void get_decimal_converter(const FieldSchema* field_schema, DataTypePtr src_logical_type,
151
                                  const DataTypePtr& dst_logical_type,
152
                                  ConvertParams* convert_params,
153
0
                                  std::unique_ptr<PhysicalToLogicalConverter>& physical_converter) {
154
0
    const tparquet::SchemaElement& parquet_schema = field_schema->parquet_schema;
155
0
    if (is_decimal(dst_logical_type->get_primitive_type())) {
156
0
        src_logical_type = create_decimal(parquet_schema.precision, parquet_schema.scale, false);
157
0
    }
158
159
0
    tparquet::Type::type src_physical_type = parquet_schema.type;
160
0
    PrimitiveType src_logical_primitive = src_logical_type->get_primitive_type();
161
162
0
    if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
163
0
        switch (src_logical_primitive) {
164
0
#define DISPATCH(LOGICAL_PTYPE)                                                     \
165
0
    case LOGICAL_PTYPE: {                                                           \
166
0
        physical_converter.reset(                                                   \
167
0
                new FixedSizeToDecimal<LOGICAL_PTYPE>(parquet_schema.type_length)); \
168
0
        break;                                                                      \
169
0
    }
170
0
            FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
171
0
#undef DISPATCH
172
0
        default:
173
0
            physical_converter =
174
0
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
175
0
        }
176
0
    } else if (src_physical_type == tparquet::Type::BYTE_ARRAY) {
177
0
        switch (src_logical_primitive) {
178
0
#define DISPATCH(LOGICAL_PTYPE)                                         \
179
0
    case LOGICAL_PTYPE: {                                               \
180
0
        physical_converter.reset(new StringToDecimal<LOGICAL_PTYPE>()); \
181
0
        break;                                                          \
182
0
    }
183
0
            FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
184
0
#undef DISPATCH
185
0
        default:
186
0
            physical_converter =
187
0
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
188
0
        }
189
0
    } else if (src_physical_type == tparquet::Type::INT32 ||
190
0
               src_physical_type == tparquet::Type::INT64) {
191
0
        switch (src_logical_primitive) {
192
0
#define DISPATCH(LOGICAL_PTYPE)                                                          \
193
0
    case LOGICAL_PTYPE: {                                                                \
194
0
        if (src_physical_type == tparquet::Type::INT32) {                                \
195
0
            physical_converter.reset(new NumberToDecimal<TYPE_INT, LOGICAL_PTYPE>());    \
196
0
        } else {                                                                         \
197
0
            physical_converter.reset(new NumberToDecimal<TYPE_BIGINT, LOGICAL_PTYPE>()); \
198
0
        }                                                                                \
199
0
        break;                                                                           \
200
0
    }
201
0
            FOR_LOGICAL_DECIMAL_TYPES(DISPATCH)
202
0
#undef DISPATCH
203
0
        default:
204
0
            physical_converter =
205
0
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
206
0
        }
207
0
    } else {
208
0
        physical_converter =
209
0
                std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
210
0
    }
211
0
}
212
213
namespace {
214
215
class NativeUUIDConverter final : public PhysicalToLogicalConverter {
216
public:
217
1
    Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override {
218
1
        const auto from_col = remove_nullable(src_physical_col);
219
1
        const auto src_data = get_fixed_length_physical_data(*from_col, UUIDValue::BINARY_LENGTH);
220
1
        auto& values =
221
1
                assert_cast<ColumnUUID&>(*get_mutable_inner_column(src_logical_column)).get_data();
222
3
        for (size_t i = 0; i < src_data.rows; ++i) {
223
            // File UUIDs use network byte order, unlike the native column's integer layout.
224
2
            values.push_back(UUIDValue::from_big_endian(reinterpret_cast<const uint8_t*>(
225
2
                    src_data.data + i * UUIDValue::BINARY_LENGTH)));
226
2
        }
227
1
        return Status::OK();
228
1
    }
229
};
230
231
class UUIDStringConverter final : public PhysicalToLogicalConverter {
232
public:
233
2
    Status physical_convert(ColumnPtr& src_physical_col, ColumnPtr& src_logical_column) override {
234
2
        const auto from_col = remove_nullable(src_physical_col);
235
2
        const auto src_data = get_fixed_length_physical_data(*from_col, 16);
236
2
        auto& strings = assert_cast<ColumnString&>(*get_mutable_inner_column(src_logical_column));
237
8
        for (size_t i = 0; i < src_data.rows; ++i) {
238
6
            const auto value = UUIDValue::from_big_endian(
239
6
                    reinterpret_cast<const uint8_t*>(src_data.data + i * 16));
240
6
            const auto text = UUIDValue::to_string(value);
241
6
            strings.insert_data(text.data(), text.size());
242
6
        }
243
2
        return Status::OK();
244
2
    }
245
};
246
247
} // namespace
248
249
std::unique_ptr<PhysicalToLogicalConverter> PhysicalToLogicalConverter::get_converter(
250
        const FieldSchema* field_schema, DataTypePtr src_logical_type,
251
        const DataTypePtr& dst_logical_type, const cctz::time_zone* ctz, bool is_dict_filter,
252
78
        bool preserve_binary_uuid) {
253
78
    std::unique_ptr<ConvertParams> convert_params = std::make_unique<ConvertParams>();
254
78
    const tparquet::SchemaElement& parquet_schema = field_schema->parquet_schema;
255
78
    convert_params->init(field_schema, ctz);
256
78
    tparquet::Type::type src_physical_type = parquet_schema.type;
257
78
    std::unique_ptr<PhysicalToLogicalConverter> physical_converter;
258
78
    if (is_dict_filter) {
259
0
        src_physical_type = tparquet::Type::INT32;
260
0
        src_logical_type = DataTypeFactory::instance().create_data_type(
261
0
                PrimitiveType::TYPE_INT, dst_logical_type->is_nullable());
262
0
    }
263
78
    PrimitiveType src_logical_primitive = src_logical_type->get_primitive_type();
264
265
78
    if (field_schema->is_type_compatibility) {
266
0
        if (src_logical_primitive == TYPE_SMALLINT) {
267
0
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_SMALLINT>>();
268
0
        } else if (src_logical_primitive == TYPE_INT) {
269
0
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_INT>>();
270
0
        } else if (src_logical_primitive == TYPE_BIGINT) {
271
0
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_BIGINT>>();
272
0
        } else if (src_logical_primitive == TYPE_LARGEINT) {
273
0
            physical_converter = std::make_unique<UnsignedIntegerConverter<TYPE_LARGEINT>>();
274
0
        } else {
275
0
            physical_converter =
276
0
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
277
0
        }
278
78
    } else if (is_parquet_native_type(src_logical_primitive)) {
279
63
        bool is_string_logical_type = is_string_type(src_logical_primitive);
280
63
        if (is_string_logical_type && src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
281
6
            if (parquet_schema.logicalType.__isset.UUID && !preserve_binary_uuid) {
282
                // Ordinary Parquet scans expose canonical text, as in File Scanner V2.
283
                // Iceberg retains its raw 16-byte STRING carrier for equality deletes/defaults.
284
2
                if (parquet_schema.type_length != UUIDValue::BINARY_LENGTH) {
285
1
                    physical_converter = std::make_unique<UnsupportedConverter>(src_physical_type,
286
1
                                                                                src_logical_type);
287
1
                } else {
288
1
                    physical_converter = std::make_unique<UUIDStringConverter>();
289
1
                }
290
4
            } else {
291
4
                physical_converter =
292
4
                        std::make_unique<FixedSizeBinaryConverter>(parquet_schema.type_length);
293
4
            }
294
57
        } else if (src_logical_primitive == TYPE_FLOAT &&
295
57
                   src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY &&
296
57
                   parquet_schema.logicalType.__isset.FLOAT16) {
297
0
            physical_converter =
298
0
                    std::make_unique<Float16PhysicalConverter>(parquet_schema.type_length);
299
57
        } else {
300
57
            physical_converter = std::make_unique<ConsistentPhysicalConverter>();
301
57
        }
302
63
    } else if (src_logical_primitive == TYPE_TINYINT) {
303
0
        physical_converter = std::make_unique<LittleIntPhysicalConverter<TYPE_TINYINT>>();
304
15
    } else if (src_logical_primitive == TYPE_SMALLINT) {
305
0
        physical_converter = std::make_unique<LittleIntPhysicalConverter<TYPE_SMALLINT>>();
306
15
    } else if (is_decimal_type(src_logical_primitive)) {
307
0
        get_decimal_converter(field_schema, src_logical_type, dst_logical_type,
308
0
                              convert_params.get(), physical_converter);
309
15
    } else if (src_logical_primitive == TYPE_DATEV2) {
310
0
        physical_converter = std::make_unique<Int32ToDate>();
311
15
    } else if (src_logical_primitive == TYPE_DATETIMEV2) {
312
11
        if (src_physical_type == tparquet::Type::INT96) {
313
            // int96 only stores nanoseconds in standard parquet file
314
1
            convert_params->reset_time_scale_if_missing(9);
315
1
            physical_converter = std::make_unique<Int96toTimestamp>();
316
10
        } else if (src_physical_type == tparquet::Type::INT64) {
317
10
            convert_params->reset_time_scale_if_missing(src_logical_type->get_scale());
318
10
            physical_converter = std::make_unique<Int64ToTimestamp>();
319
10
        } else {
320
0
            physical_converter =
321
0
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
322
0
        }
323
11
    } else if (src_logical_primitive == TYPE_UUID) {
324
2
        if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY &&
325
2
            std::cmp_equal(parquet_schema.type_length, UUIDValue::BINARY_LENGTH)) {
326
1
            physical_converter = std::make_unique<NativeUUIDConverter>();
327
1
        } else {
328
1
            physical_converter =
329
1
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
330
1
        }
331
2
    } else if (src_logical_primitive == TYPE_VARBINARY) {
332
0
        if (src_physical_type == tparquet::Type::FIXED_LEN_BYTE_ARRAY) {
333
0
            DCHECK(parquet_schema.logicalType.__isset.UUID) << parquet_schema.name;
334
0
            physical_converter =
335
0
                    std::make_unique<UUIDVarBinaryConverter>(parquet_schema.type_length);
336
0
        } else {
337
0
            DCHECK(src_physical_type == tparquet::Type::BYTE_ARRAY) << src_physical_type;
338
0
            physical_converter = std::make_unique<ConsistentPhysicalConverter>();
339
0
        }
340
2
    } else if (src_logical_primitive == TYPE_TIMESTAMPTZ) {
341
2
        if (src_physical_type == tparquet::Type::INT96) {
342
1
            physical_converter = std::make_unique<Int96toTimestampTz>();
343
1
        } else if (src_physical_type == tparquet::Type::INT64) {
344
1
            DCHECK(src_physical_type == tparquet::Type::INT64) << src_physical_type;
345
1
            DCHECK(parquet_schema.logicalType.__isset.TIMESTAMP) << parquet_schema.name;
346
1
            physical_converter = std::make_unique<Int64ToTimestampTz>();
347
1
        } else {
348
0
            physical_converter =
349
0
                    std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
350
0
        }
351
2
    } else {
352
0
        physical_converter =
353
0
                std::make_unique<UnsupportedConverter>(src_physical_type, src_logical_type);
354
0
    }
355
356
78
    if (physical_converter->support()) {
357
76
        physical_converter->_convert_params = std::move(convert_params);
358
76
        physical_converter->_logical_converter = converter::ColumnTypeConverter::get_converter(
359
76
                src_logical_type, dst_logical_type, converter::FileFormat::PARQUET);
360
76
        if (!physical_converter->_logical_converter->support()) {
361
0
            physical_converter = std::make_unique<UnsupportedConverter>(
362
0
                    "Unsupported type change: " +
363
0
                    physical_converter->_logical_converter->get_error_msg());
364
0
        }
365
76
    }
366
78
    return physical_converter;
367
78
}
368
369
} // namespace doris::parquet