be/src/format/parquet/decoder.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/decoder.h"

#include <cctz/time_zone.h>
#include <gen_cpp/parquet_types.h>

#include "format/parquet/bool_plain_decoder.h"
#include "format/parquet/bool_rle_decoder.h"
#include "format/parquet/byte_array_dict_decoder.h"
#include "format/parquet/byte_array_plain_decoder.h"
#include "format/parquet/byte_stream_split_decoder.h"
#include "format/parquet/delta_bit_pack_decoder.h"
#include "format/parquet/fix_length_dict_decoder.hpp"
#include "format/parquet/fix_length_plain_decoder.h"

namespace doris {
Status Decoder::get_decoder(tparquet::Type::type type, tparquet::Encoding::type encoding,
                            std::unique_ptr<Decoder>& decoder) {
    switch (encoding) {
    case tparquet::Encoding::PLAIN:
        switch (type) {
        case tparquet::Type::BOOLEAN:
            decoder.reset(new BoolPlainDecoder());
            break;
        case tparquet::Type::BYTE_ARRAY:
            decoder.reset(new ByteArrayPlainDecoder());
            break;
        case tparquet::Type::INT32:
        case tparquet::Type::INT64:
        case tparquet::Type::INT96:
        case tparquet::Type::FLOAT:
        case tparquet::Type::DOUBLE:
        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
            decoder.reset(new FixLengthPlainDecoder());
            break;
        default:
            return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
                                         tparquet::to_string(type), tparquet::to_string(encoding));
        }
        break;
    case tparquet::Encoding::RLE_DICTIONARY:
        switch (type) {
        case tparquet::Type::BOOLEAN:
            return Status::InternalError("Bool type can't has dictionary page");
        case tparquet::Type::BYTE_ARRAY:
            decoder.reset(new ByteArrayDictDecoder());
            break;
        case tparquet::Type::INT32:
            decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT32>());
            break;
        case tparquet::Type::INT64:
            decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT64>());
            break;
        case tparquet::Type::INT96:
            decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT96>());
            break;
        case tparquet::Type::FLOAT:
            decoder.reset(new FixLengthDictDecoder<tparquet::Type::FLOAT>());
            break;
        case tparquet::Type::DOUBLE:
            decoder.reset(new FixLengthDictDecoder<tparquet::Type::DOUBLE>());
            break;
        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
            decoder.reset(new FixLengthDictDecoder<tparquet::Type::FIXED_LEN_BYTE_ARRAY>());
            break;
        default:
            return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
                                         tparquet::to_string(type), tparquet::to_string(encoding));
        }
        break;
    case tparquet::Encoding::RLE:
        switch (type) {
        case tparquet::Type::BOOLEAN:
            decoder.reset(new BoolRLEDecoder());
            break;
        default:
            return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
                                         tparquet::to_string(type), tparquet::to_string(encoding));
        }
        break;
    case tparquet::Encoding::DELTA_BINARY_PACKED:
        // Supports only INT32 and INT64.
        switch (type) {
        case tparquet::Type::INT32:
            decoder.reset(new DeltaBitPackDecoder<int32_t>());
            break;
        case tparquet::Type::INT64:
            decoder.reset(new DeltaBitPackDecoder<int64_t>());
            break;
        default:
            return Status::InternalError("DELTA_BINARY_PACKED only supports INT32 and INT64");
        }
        break;
    case tparquet::Encoding::DELTA_BYTE_ARRAY:
        switch (type) {
        case tparquet::Type::BYTE_ARRAY:
        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
            decoder.reset(new DeltaByteArrayDecoder());
            break;
        default:
            return Status::InternalError(
                    "DELTA_BYTE_ARRAY only supports BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY.");
        }
        break;
    case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
        switch (type) {
        case tparquet::Type::BYTE_ARRAY:
            decoder.reset(new DeltaLengthByteArrayDecoder());
            break;
        default:
            return Status::InternalError("DELTA_LENGTH_BYTE_ARRAY only supports BYTE_ARRAY.");
        }
        break;
    case tparquet::Encoding::BYTE_STREAM_SPLIT:
        switch (type) {
        case tparquet::Type::INT32:
        case tparquet::Type::INT64:
        case tparquet::Type::INT96:
        case tparquet::Type::FLOAT:
        case tparquet::Type::DOUBLE:
        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
            decoder.reset(new ByteStreamSplitDecoder());
            break;
        default:
            return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
                                         tparquet::to_string(type), tparquet::to_string(encoding));
        }
        break;
    default:
        return Status::InternalError("Unsupported encoding {}(type={}) in parquet decoder",
                                     tparquet::to_string(encoding), tparquet::to_string(type));
    }
    return Status::OK();
}

} // namespace doris

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/decoder.h"
19
20		#include <cctz/time_zone.h>
21		#include <gen_cpp/parquet_types.h>
22
23		#include "format/parquet/bool_plain_decoder.h"
24		#include "format/parquet/bool_rle_decoder.h"
25		#include "format/parquet/byte_array_dict_decoder.h"
26		#include "format/parquet/byte_array_plain_decoder.h"
27		#include "format/parquet/byte_stream_split_decoder.h"
28		#include "format/parquet/delta_bit_pack_decoder.h"
29		#include "format/parquet/fix_length_dict_decoder.hpp"
30		#include "format/parquet/fix_length_plain_decoder.h"
31
32		namespace doris {
33		Status Decoder::get_decoder(tparquet::Type::type type, tparquet::Encoding::type encoding,
34	8.51k	std::unique_ptr<Decoder>& decoder) {
35	8.51k	switch (encoding) {
36	1.28k	case tparquet::Encoding::PLAIN:
37	1.28k	switch (type) {
38	298	case tparquet::Type::BOOLEAN:
39	298	decoder.reset(new BoolPlainDecoder());
40	298	break;
41	323	case tparquet::Type::BYTE_ARRAY:
42	323	decoder.reset(new ByteArrayPlainDecoder());
43	323	break;
44	412	case tparquet::Type::INT32:
45	490	case tparquet::Type::INT64:
46	502	case tparquet::Type::INT96:
47	554	case tparquet::Type::FLOAT:
48	609	case tparquet::Type::DOUBLE:
49	666	case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
50	666	decoder.reset(new FixLengthPlainDecoder());
51	666	break;
52	0	default:
53	0	return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
54	0	tparquet::to_string(type), tparquet::to_string(encoding));
55	1.28k	}
56	1.28k	break;
57	7.02k	case tparquet::Encoding::RLE_DICTIONARY:
58	7.02k	switch (type) {
59	0	case tparquet::Type::BOOLEAN:
60	0	return Status::InternalError("Bool type can't has dictionary page");
61	2.16k	case tparquet::Type::BYTE_ARRAY:
62	2.16k	decoder.reset(new ByteArrayDictDecoder());
63	2.16k	break;
64	2.68k	case tparquet::Type::INT32:
65	2.68k	decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT32>());
66	2.68k	break;
67	851	case tparquet::Type::INT64:
68	851	decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT64>());
69	851	break;
70	76	case tparquet::Type::INT96:
71	76	decoder.reset(new FixLengthDictDecoder<tparquet::Type::INT96>());
72	76	break;
73	145	case tparquet::Type::FLOAT:
74	145	decoder.reset(new FixLengthDictDecoder<tparquet::Type::FLOAT>());
75	145	break;
76	896	case tparquet::Type::DOUBLE:
77	896	decoder.reset(new FixLengthDictDecoder<tparquet::Type::DOUBLE>());
78	896	break;
79	208	case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
80	208	decoder.reset(new FixLengthDictDecoder<tparquet::Type::FIXED_LEN_BYTE_ARRAY>());
81	208	break;
82	0	default:
83	0	return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
84	0	tparquet::to_string(type), tparquet::to_string(encoding));
85	7.02k	}
86	7.02k	break;
87	7.02k	case tparquet::Encoding::RLE:
88	0	switch (type) {
89	0	case tparquet::Type::BOOLEAN:
90	0	decoder.reset(new BoolRLEDecoder());
91	0	break;
92	0	default:
93	0	return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
94	0	tparquet::to_string(type), tparquet::to_string(encoding));
95	0	}
96	0	break;
97	195	case tparquet::Encoding::DELTA_BINARY_PACKED:
98		// Supports only INT32 and INT64.
99	195	switch (type) {
100	0	case tparquet::Type::INT32:
101	0	decoder.reset(new DeltaBitPackDecoder<int32_t>());
102	0	break;
103	195	case tparquet::Type::INT64:
104	195	decoder.reset(new DeltaBitPackDecoder<int64_t>());
105	195	break;
106	0	default:
107	0	return Status::InternalError("DELTA_BINARY_PACKED only supports INT32 and INT64");
108	195	}
109	195	break;
110	195	case tparquet::Encoding::DELTA_BYTE_ARRAY:
111	0	switch (type) {
112	0	case tparquet::Type::BYTE_ARRAY:
113	0	case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
114	0	decoder.reset(new DeltaByteArrayDecoder());
115	0	break;
116	0	default:
117	0	return Status::InternalError(
118	0	"DELTA_BYTE_ARRAY only supports BYTE_ARRAY, FIXED_LEN_BYTE_ARRAY.");
119	0	}
120	0	break;
121	0	case tparquet::Encoding::DELTA_LENGTH_BYTE_ARRAY:
122	0	switch (type) {
123	0	case tparquet::Type::BYTE_ARRAY:
124	0	decoder.reset(new DeltaLengthByteArrayDecoder());
125	0	break;
126	0	default:
127	0	return Status::InternalError("DELTA_LENGTH_BYTE_ARRAY only supports BYTE_ARRAY.");
128	0	}
129	0	break;
130	0	case tparquet::Encoding::BYTE_STREAM_SPLIT:
131	0	switch (type) {
132	0	case tparquet::Type::INT32:
133	0	case tparquet::Type::INT64:
134	0	case tparquet::Type::INT96:
135	0	case tparquet::Type::FLOAT:
136	0	case tparquet::Type::DOUBLE:
137	0	case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
138	0	decoder.reset(new ByteStreamSplitDecoder());
139	0	break;
140	0	default:
141	0	return Status::InternalError("Unsupported type {}(encoding={}) in parquet decoder",
142	0	tparquet::to_string(type), tparquet::to_string(encoding));
143	0	}
144	0	break;
145	0	default:
146	0	return Status::InternalError("Unsupported encoding {}(type={}) in parquet decoder",
147	0	tparquet::to_string(encoding), tparquet::to_string(type));
148	8.51k	}
149	8.50k	return Status::OK();
150	8.51k	}
151
152		} // namespace doris

Coverage Report

Created: 2026-07-04 02:13