Skip to content

branch-4.2: [feature](file) Support FILE type #70221

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
linrrzqqq wants to merge 1 commit into
base: branch-4.2
Choose a base branch
from
+13,700 −167
Open

branch-4.2: [feature](file) Support FILE type #70221

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
470 changes: 470 additions & 0 deletions be/src/core/column/column_file.cpp
View file
Open in desktop

Large diffs are not rendered by default.

126 changes: 126 additions & 0 deletions be/src/core/column/column_file.h
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
// 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 <array>

#include "core/column/column.h"
#include "core/cow.h"

namespace doris {

// Complete physical FILE values: uri, offset, size, content_type, checksum, inline.
// All six children are nullable; parent SQL NULL belongs to an outer ColumnNullable.
// Input-boundary validation is responsible for URI/range/metadata validity.
class ColumnFile final : public COWHelper<IColumn, ColumnFile> {
private:
friend class COWHelper<IColumn, ColumnFile>;

ColumnFile();
explicit ColumnFile(MutableColumns&& columns);
ColumnFile(const ColumnFile&) = default;

public:
static constexpr size_t NUM_CHILDREN = 6;
using FileColumns = std::array<IColumn::WrappedPtr, NUM_CHILDREN>;

std::string get_name() const override { return "File"; }
size_t size() const override { return _columns[0]->size(); }
size_t tuple_size() const { return NUM_CHILDREN; }
bool is_variable_length() const override { return true; }
bool is_exclusive() const override {
return IColumn::is_exclusive() &&
std::all_of(_columns.begin(), _columns.end(),
[](const auto& column) { return column->is_exclusive(); });
}
void sanity_check() const override;
const IColumn& get_column(size_t i) const { return *_columns[i]; }
IColumn& get_column(size_t i) { return *_columns[i]; }
const FileColumns& get_columns() const { return _columns; }
Columns get_columns_copy() const { return {_columns.begin(), _columns.end()}; }
const ColumnPtr& get_column_ptr(size_t i) const { return _columns[i]; }
ColumnPtr& get_column_ptr(size_t i) { return _columns[i]; }

MutableColumnPtr clone_resized(size_t size) const override;
Field operator[](size_t n) const override;
void get(size_t n, Field& res) const override;
void insert(const Field& value) override;
void insert_from(const IColumn& src, size_t n) override;
void insert_range_from(const IColumn& src, size_t start, size_t length) override;
void insert_range_from_ignore_overflow(const IColumn& src, size_t start,
size_t length) override;
void insert_indices_from(const IColumn& src, const uint32_t* begin,
const uint32_t* end) override;
void insert_many_from(const IColumn& src, size_t position, size_t length) override;
void replace_column_data(const IColumn& rhs, size_t row, size_t self_row = 0) override;
void insert_default() override;
void insert_many_defaults(size_t length) override;
void pop_back(size_t n) override;
void reserve(size_t n) override;
void resize(size_t n) override;
void clear() override;
void erase(size_t start, size_t length) override;
ColumnPtr filter(const Filter& filter, ssize_t result_size_hint) const override;
size_t filter(const Filter& filter) override;
Status filter_by_selector(const uint16_t* selector, size_t count,
IColumn* destination) const override;
MutableColumnPtr permute(const Permutation& permutation, size_t limit) const override;
ColumnPtr convert_column_if_overflow() override;

size_t byte_size() const override;
size_t allocated_bytes() const override;
bool has_enough_capacity(const IColumn& src) const override;
void for_each_subcolumn(ColumnCallback callback) override;

StringRef serialize_value_into_arena(size_t n, Arena& arena, const char*& begin) const override;
const char* deserialize_and_insert_from_arena(const char* pos) override;
size_t serialize_size_at(size_t row) const override;
size_t serialize_impl(char* pos, size_t row) const override;
size_t deserialize_impl(const char* pos) override;
void serialize_vec(StringRef* keys, size_t num_rows) const override;
void deserialize_vec(StringRef* keys, size_t num_rows) override;
size_t get_max_row_byte_size() const override;

// These hashes cover serialized physical values for internal integrity only.
// They do not grant FILE SQL equality, hashing, or deduplication semantics.
void update_hash_with_value(size_t n, SipHash& hash) const override;
void update_hashes_with_value(uint64_t* __restrict hashes,
const uint8_t* __restrict null_data = nullptr) const override;
void update_xxHash_with_value(size_t start, size_t end, uint64_t& hash,
const uint8_t* __restrict null_data) const override;
void update_crcs_with_value(uint32_t* __restrict hashes, PrimitiveType type, uint32_t rows,
uint32_t offset = 0,
const uint8_t* __restrict null_data = nullptr) const override;
void update_crc_with_value(size_t start, size_t end, uint32_t& hash,
const uint8_t* __restrict null_data) const override;
void update_crc32c_batch(uint32_t* __restrict hashes,
const uint8_t* __restrict null_map) const override;
void update_crc32c_single(size_t start, size_t end, uint32_t& hash,
const uint8_t* __restrict null_map) const override;

int compare_at(size_t n, size_t m, const IColumn& rhs, int nan_direction_hint) const override;
void get_permutation(bool reverse, size_t limit, int nan_direction_hint, HybridSorter& sorter,
Permutation& res) const override;
void sort_column(const ColumnSorter* sorter, EqualFlags& flags, Permutation& permutation,
EqualRange& range, bool last_column) const override;

private:
FileColumns _columns;
};

} // namespace doris
2 changes: 2 additions & 0 deletions be/src/core/data_type/data_type.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,8 @@ PGenericType_TypeId IDataType::get_pdata_type(const IDataType* data_type) {
return PGenericType::LIST;
case PrimitiveType::TYPE_STRUCT:
return PGenericType::STRUCT;
case PrimitiveType::TYPE_FILE:
return PGenericType::FILE;
case PrimitiveType::TYPE_JSONB:
return PGenericType::JSONB;
case PrimitiveType::TYPE_MAP:
Expand Down
78 changes: 75 additions & 3 deletions be/src/core/data_type/data_type_factory.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
#include "core/data_type/data_type_date_or_datetime_v2.h"
#include "core/data_type/data_type_date_time.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_file.h"
#include "core/data_type/data_type_fixed_length_object.h"
#include "core/data_type/data_type_hll.h"
#include "core/data_type/data_type_ipv4.h"
Expand Down Expand Up @@ -327,6 +328,16 @@ DataTypePtr DataTypeFactory::create_data_type(const PColumnMeta& pcolumn) {
nested = std::make_shared<DataTypeMap>(create_data_type(pcolumn.children(0)),
create_data_type(pcolumn.children(1)));
break;
case PGenericType::FILE: {
DataTypes children;
Strings names;
for (const auto& child : pcolumn.children()) {
children.push_back(create_data_type(child));
names.push_back(child.name());
}
nested = std::make_shared<DataTypeFile>(children, names);
break;
}
case PGenericType::STRUCT: {
int col_size = pcolumn.children_size();
DCHECK(col_size >= 1);
Expand Down Expand Up @@ -512,6 +523,9 @@ DataTypePtr DataTypeFactory::create_data_type(const PrimitiveType primitive_type
const_cast<DataTypeUInt8&>(reinterpret_cast<const DataTypeUInt8&>(*nested))
.set_null_literal(true);
break;
case TYPE_FILE:
nested = std::make_shared<DataTypeFile>();
break;
case TYPE_VARBINARY:
nested = std::make_shared<DataTypeVarbinary>(len, TYPE_VARBINARY);
break;
Expand Down Expand Up @@ -560,6 +574,8 @@ DataTypePtr DataTypeFactory::create_data_type(const std::vector<TTypeNode>& type
doc_mode);
}
return is_nullable ? make_nullable(dt) : dt;
} else if (scalar_type.type == TPrimitiveType::FILE) {
throw Exception(ErrorCode::INVALID_ARGUMENT, "FILE requires a FILE type node");
}
return create_data_type(thrift_to_type(scalar_type.type), is_nullable,
scalar_type.__isset.precision ? scalar_type.precision : 0,
Expand All @@ -577,7 +593,26 @@ DataTypePtr DataTypeFactory::create_data_type(const std::vector<TTypeNode>& type
create_data_type(types, idx, node.contains_nulls[0]));
break;
}
case TTypeNodeType::FILE:
case TTypeNodeType::STRUCT: {
if (node.type == TTypeNodeType::FILE &&
(node.struct_fields.size() != DataTypeFile::FIELD_COUNT ||
types.size() - static_cast<size_t>(*idx) <= DataTypeFile::FIELD_COUNT)) {
throw Exception(ErrorCode::INVALID_ARGUMENT, "Invalid FILE type descriptor");
}
if (node.type == TTypeNodeType::FILE) {
constexpr TPrimitiveType::type child_types[] = {
TPrimitiveType::VARCHAR, TPrimitiveType::BIGINT, TPrimitiveType::BIGINT,
TPrimitiveType::VARCHAR, TPrimitiveType::VARCHAR, TPrimitiveType::VARBINARY};
for (size_t i = 0; i < DataTypeFile::FIELD_COUNT; ++i) {
const auto& child = types[*idx + i + 1];
if (child.type != TTypeNodeType::SCALAR || !child.__isset.scalar_type ||
child.scalar_type.type != child_types[i]) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"FILE requires its canonical scalar children");
}
}
}
DCHECK(!node.__isset.scalar_type);
DCHECK_LT(*idx, types.size() - 1);
DCHECK(!node.__isset.contains_nulls);
Expand All @@ -592,7 +627,11 @@ DataTypePtr DataTypeFactory::create_data_type(const std::vector<TTypeNode>& type
data_types.push_back(create_data_type(types, idx, node.struct_fields[i].contains_null));
names.push_back(node.struct_fields[i].name);
}
nested = std::make_shared<DataTypeStruct>(data_types, names);
if (node.type == TTypeNodeType::FILE) {
nested = std::make_shared<DataTypeFile>(data_types, names);
} else {
nested = std::make_shared<DataTypeStruct>(data_types, names);
}
break;
}
case TTypeNodeType::MAP: {
Expand Down Expand Up @@ -673,6 +712,8 @@ DataTypePtr DataTypeFactory::create_data_type(
nested = std::make_shared<DataTypeVariant>(node.variant_max_subcolumns_count(),
node.variant_enable_doc_mode());
}
} else if (primitive_type == TYPE_FILE) {
throw Exception(ErrorCode::INVALID_ARGUMENT, "FILE requires a FILE type node");
} else {
return create_data_type(primitive_type, is_nullable,
scalar_type.has_precision() ? scalar_type.precision() : 0,
Expand All @@ -698,18 +739,49 @@ DataTypePtr DataTypeFactory::create_data_type(
nested = std::make_shared<DataTypeMap>(data_types[0], data_types[1]);
break;
}
case TTypeNodeType::FILE:
case TTypeNodeType::STRUCT: {
if (node.type() == TTypeNodeType::FILE &&
(node.struct_fields_size() != DataTypeFile::FIELD_COUNT ||
types.size() - *idx <= DataTypeFile::FIELD_COUNT)) {
throw Exception(ErrorCode::INVALID_ARGUMENT, "Invalid FILE protobuf type descriptor");
}
if (node.type() == TTypeNodeType::FILE) {
constexpr TPrimitiveType::type child_types[] = {
TPrimitiveType::VARCHAR, TPrimitiveType::BIGINT, TPrimitiveType::BIGINT,
TPrimitiveType::VARCHAR, TPrimitiveType::VARCHAR, TPrimitiveType::VARBINARY};
for (int i = 0; i < DataTypeFile::FIELD_COUNT; ++i) {
const auto& child = types.Get(*idx + i + 1);
if (child.type() != TTypeNodeType::SCALAR || !child.has_scalar_type() ||
child.scalar_type().type() != child_types[i]) {
throw Exception(ErrorCode::INVALID_ARGUMENT,
"FILE requires its canonical scalar children");
}
}
}
DataTypes data_types;
Strings names;
data_types.reserve(node.struct_fields_size());
names.reserve(node.struct_fields_size());
for (size_t i = 0; i < node.struct_fields_size(); i++) {
const auto& field = node.struct_fields(cast_set<int>(i));
++(*idx);
data_types.push_back(create_data_type(types, idx, field.contains_null()));
if (node.type() == TTypeNodeType::FILE) {
const auto& scalar_type = types.Get(*idx).scalar_type();
data_types.push_back(create_data_type(
thrift_to_type(static_cast<TPrimitiveType::type>(scalar_type.type())),
field.contains_null(), 0, 0,
scalar_type.has_len() ? scalar_type.len() : -1));
} else {
data_types.push_back(create_data_type(types, idx, field.contains_null()));
}
names.push_back(field.name());
}
nested = std::make_shared<DataTypeStruct>(data_types, names);
if (node.type() == TTypeNodeType::FILE) {
nested = std::make_shared<DataTypeFile>(data_types, names);
} else {
nested = std::make_shared<DataTypeStruct>(data_types, names);
}
break;
}
case TTypeNodeType::VARIANT: {
Expand Down
Loading
Loading