github-actions[bot] commented on code in PR #69040:
URL: https://github.com/apache/doris/pull/69040#discussion_r4374374677
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/util/ExpressionUtils.java:
##########
@@ -184,6 +185,27 @@ private static boolean isUnsafeDateArithmetic(Expression
expression) {
return false;
}
+ /** Reject FILE values in SQL keys that require comparison, ordering, or
hashing. */
+ public static void checkFileKey(Expression value, String operation) {
+ if (value.getDataType().typeContainsFile()) {
+ throw new AnalysisException(operation + " does not support FILE or
a type containing FILE: "
+ + value.getDataType().toSql());
+ }
+ }
+
+ /** Called by the existing recursive expression type check, without a
second plan traversal. */
+ public static void checkFileExpression(Expression expression) {
Review Comment:
[P2] Check FILE keys inside aggregate ORDER BY expressions
This reduced plan is admitted without a LogicalSort or WindowExpression:
```text
Aggregate(output=[GROUP_CONCAT(CAST(number AS STRING) ORDER BY f)])
Project(number, CAST(NAMED_STRUCT('uri', CONCAT('urn:',number),
'offset',NULL,'size',NULL,'content_type',NULL,
'checksum',NULL,'inline',NULL) AS FILE) AS f)
Numbers(number=2)
```
`GroupConcat` accepts ANY for its OrderExpression, while this recursive
check ignores that expression shape. `translateAggregateFunction` preserves the
FILE order argument; BE installs `AggregateFunctionSort`, and
`AggregateFunctionSortData::sort` calls `sort_block` ->
`ColumnFile::get_permutation`, which throws `FILE sorting is not supported`.
Apply the recursive `checkFileKey` restriction to OrderExpression children and
cover nested FILE keys too. This is a separate ordering path from the existing
aggregate-DISTINCT report: the trigger has no DISTINCT or deduplication key.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/CreateFunctionCommand.java:
##########
@@ -445,6 +439,9 @@ private void checkUdfSupportedType(Type type, String
typePosition) throws Analys
for (StructField field : ((StructType) type).getFields()) {
checkUdfSupportedType(field.getType(), typePosition + " field
" + field.getName());
}
+ } else if (type.isObjectStored()) {
Review Comment:
[P2] Provide transport for the admitted Python FILE signatures
This accepts a Python function declared as `py(FILE) RETURNS INT` (the new
CREATE tests also assert acceptance of nested FILE arguments/returns). Calling
`py(f)` with a valid nonconstant six-field STRUCT-to-FILE value cannot execute:
`PythonFunctionCall::open` passes FILE into `PythonUDFMeta::serialize_to_json`,
whose `convert_types_to_schema` calls
`DorisArrowSchemaConvertor::convert_to_arrow_type`. That switch has no FILE
case and returns `Unknown primitive type(File) convert to Arrow type`;
ARRAY/MAP/STRUCT recurse into the same failure. Both FILE Arrow serde
directions explicitly return NotSupported as well. The scalar INT result passes
the FILE result-sink fence, so it does not prevent this path. Implement the
Python FILE schema/serde bridge, or recursively reject unsupported signatures
at CREATE, and add a CREATE-and-invoke case instead of validating catalog
acceptance alone.
##########
be/src/core/column/column_file.cpp:
##########
@@ -0,0 +1,470 @@
+// 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 "core/column/column_file.h"
+
+#include <crc32c/crc32c.h>
+
+#include <algorithm>
+#include <limits>
+
+#include "core/arena.h"
+#include "core/assert_cast.h"
+#include "core/column/column_nullable.h"
+#include "core/column/column_string.h"
+#include "core/column/column_varbinary.h"
+#include "core/column/column_vector.h"
+#include "core/column/columns_common.h"
+#include "core/custom_allocator.h"
+#include "core/data_type/primitive_type.h"
+#include "core/field.h"
+#include "exec/common/sip_hash.h"
+#include "util/hash_util.hpp"
+
+namespace doris {
+namespace {
+
+MutableColumns make_file_children() {
+ MutableColumns children;
+ children.reserve(ColumnFile::NUM_CHILDREN);
+ children.push_back(ColumnNullable::create(ColumnString::create(),
ColumnUInt8::create()));
+ children.push_back(ColumnNullable::create(ColumnInt64::create(),
ColumnUInt8::create()));
+ children.push_back(ColumnNullable::create(ColumnInt64::create(),
ColumnUInt8::create()));
+ children.push_back(ColumnNullable::create(ColumnString::create(),
ColumnUInt8::create()));
+ children.push_back(ColumnNullable::create(ColumnString::create(),
ColumnUInt8::create()));
+ children.push_back(ColumnNullable::create(ColumnVarbinary::create(),
ColumnUInt8::create()));
+ return children;
+}
+
+// VARBINARY has no column hash implementation. Hash the lossless row encoding,
+// which includes every child null flag and length, without exposing SQL
hashing.
+template <typename Update>
+void hash_file_rows(const ColumnFile& column, size_t start, size_t end, const
uint8_t* null_map,
+ Update&& update) {
+ DorisVector<char> buffer;
+ for (size_t row = start; row < end; ++row) {
+ if (null_map != nullptr && null_map[row] != 0) {
+ continue;
+ }
+ buffer.resize(column.serialize_size_at(row));
+ column.serialize_impl(buffer.data(), row);
+ update(row, buffer.data(), buffer.size());
+ }
+}
+
+uint32_t file_crc(const char* data, size_t size, uint32_t hash) {
+ // A complete row can exceed the uint32 length accepted by zlib_crc_hash.
+ while (size != 0) {
+ const auto length =
+ static_cast<uint32_t>(std::min<size_t>(size,
std::numeric_limits<uint32_t>::max()));
+ hash = HashUtil::zlib_crc_hash(data, length, hash);
+ data += length;
+ size -= length;
+ }
+ return hash;
+}
+
+} // namespace
+
+ColumnFile::ColumnFile() : ColumnFile(make_file_children()) {}
+
+ColumnFile::ColumnFile(MutableColumns&& columns) {
+ if (columns.size() != NUM_CHILDREN) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "ColumnFile requires six
children, got {}",
+ columns.size());
+ }
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i] = std::move(columns[i]);
+ }
+ sanity_check();
+ check_const_only_in_top_level();
+}
+
+void ColumnFile::sanity_check() const {
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ const auto* child = dynamic_cast<const
ColumnNullable*>(_columns[i].get());
+ if (child == nullptr) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "ColumnFile child {}
must be nullable", i);
+ }
+ const auto& nested = child->get_nested_column();
+ bool expected_type;
+ if (i == 1 || i == 2) {
+ expected_type = dynamic_cast<const ColumnInt64*>(&nested) !=
nullptr;
+ } else if (i == 5) {
+ expected_type = dynamic_cast<const ColumnVarbinary*>(&nested) !=
nullptr;
+ } else {
+ // Both string offset widths represent the same physical child
type.
+ expected_type = nested.is_column_string();
+ }
+ if (!expected_type) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "Invalid ColumnFile
child {} type {}", i,
+ nested.get_name());
+ }
+ child->sanity_check();
+ if (i != 0 && child->size() != size()) {
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "ColumnFile child {} has {} rows, expected {}", i,
child->size(),
+ size());
+ }
+ }
+}
+
+MutableColumnPtr ColumnFile::clone_resized(size_t size) const {
+ MutableColumns children;
+ children.reserve(NUM_CHILDREN);
+ for (const auto& column : _columns) {
+ children.push_back(column->clone_resized(size));
+ }
+ return create(std::move(children));
+}
+
+Field ColumnFile::operator[](size_t n) const {
+ Field result;
+ get(n, result);
+ return result;
+}
+
+void ColumnFile::get(size_t n, Field& res) const {
+ File value;
+ value.reserve(NUM_CHILDREN);
+ for (const auto& column : _columns) {
+ value.push_back((*column)[n]);
+ }
+ res = Field::create_field<TYPE_FILE>(std::move(value));
+}
+
+void ColumnFile::insert(const Field& value) {
+ DCHECK_EQ(value.get_type(), TYPE_FILE);
+ const auto& fields = value.get<TYPE_FILE>();
+ if (fields.size() != NUM_CHILDREN) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "FILE Field requires six
children, got {}",
+ fields.size());
+ }
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i]->insert(fields[i]);
+ }
+}
+
+void ColumnFile::insert_from(const IColumn& src, size_t n) {
+ const auto& file = assert_cast<const ColumnFile&>(src);
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i]->insert_from(file.get_column(i), n);
+ }
+}
+
+void ColumnFile::insert_range_from(const IColumn& src, size_t start, size_t
length) {
+ const auto& file = assert_cast<const ColumnFile&>(src);
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i]->insert_range_from(file.get_column(i), start, length);
+ }
+}
+
+void ColumnFile::insert_range_from_ignore_overflow(const IColumn& src, size_t
start,
+ size_t length) {
+ const auto& file = assert_cast<const ColumnFile&>(src);
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i]->insert_range_from_ignore_overflow(file.get_column(i),
start, length);
+ }
+}
+
+void ColumnFile::insert_indices_from(const IColumn& src, const uint32_t* begin,
+ const uint32_t* end) {
+ const auto& file = assert_cast<const ColumnFile&>(src);
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i]->insert_indices_from(file.get_column(i), begin, end);
+ }
+}
+
+void ColumnFile::insert_many_from(const IColumn& src, size_t position, size_t
length) {
+ const auto& file = assert_cast<const ColumnFile&>(src);
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ _columns[i]->insert_many_from(file.get_column(i), position, length);
+ }
+}
+
+void ColumnFile::replace_column_data(const IColumn&, size_t, size_t) {
+ // Like ColumnString/ColumnStruct, variable-length rows are replaced by
rebuilding columns.
+ throw Exception(ErrorCode::INTERNAL_ERROR,
+ "Method replace_column_data is not supported for
ColumnFile");
+}
+
+void ColumnFile::insert_default() {
+ for (auto& column : _columns) {
+ column->insert_default();
+ }
+}
+
+void ColumnFile::insert_many_defaults(size_t length) {
+ for (auto& column : _columns) {
+ column->insert_many_defaults(length);
+ }
+}
+
+void ColumnFile::pop_back(size_t n) {
+ for (auto& column : _columns) {
+ column->pop_back(n);
+ }
+}
+
+void ColumnFile::reserve(size_t n) {
+ for (auto& column : _columns) {
+ column->reserve(n);
+ }
+}
+
+void ColumnFile::resize(size_t n) {
+ if (n > size()) {
+ // ColumnNullable::resize alone does not initialize new null-map
entries.
+ insert_many_defaults(n - size());
+ } else {
+ for (auto& column : _columns) {
+ column->resize(n);
+ }
+ }
+}
+
+void ColumnFile::clear() {
+ for (auto& column : _columns) {
+ column->clear();
+ }
+}
+
+void ColumnFile::erase(size_t start, size_t length) {
+ DCHECK_LE(start, size());
+ DCHECK_LE(length, size() - start);
+ // Filtering also works for the arena-backed VARBINARY child, which has no
erase().
+ Filter keep(size(), 1);
+ std::fill(keep.begin() + start, keep.begin() + start + length, 0);
+ filter(keep);
+}
+
+ColumnPtr ColumnFile::filter(const Filter& filter, ssize_t result_size_hint)
const {
+ MutableColumns children;
+ children.reserve(NUM_CHILDREN);
+ for (const auto& column : _columns) {
+ children.push_back(IColumn::mutate(column->filter(filter,
result_size_hint)));
+ }
+ return create(std::move(children));
+}
+
+size_t ColumnFile::filter(const Filter& filter) {
+ column_match_filter_size(size(), filter.size());
+ const size_t result_size = _columns[0]->filter(filter);
+ for (size_t i = 1; i < NUM_CHILDREN - 1; ++i) {
+ const size_t child_size = _columns[i]->filter(filter);
+ DCHECK_EQ(child_size, result_size);
+ }
+ auto& inline_column = assert_cast<ColumnNullable&>(*_columns[NUM_CHILDREN
- 1]);
+ if (result_size == 0) {
+ inline_column.clear();
+ return 0;
+ }
+ // The arena remains owned by this column. Compact views without copying
large payloads.
+ auto& data =
assert_cast<ColumnVarbinary&>(inline_column.get_nested_column()).get_data();
+ size_t output_row = 0;
+ for (size_t row = 0; row < filter.size(); ++row) {
+ if (filter[row]) {
+ data[output_row++] = data[row];
+ }
+ }
+ data.resize(output_row);
+ const size_t null_map_size =
inline_column.get_null_map_column().filter(filter);
+ DCHECK_EQ(output_row, result_size);
+ DCHECK_EQ(null_map_size, result_size);
+ return result_size;
+}
+
+Status ColumnFile::filter_by_selector(const uint16_t* selector, size_t count,
+ IColumn* destination) const {
+ auto& output = assert_cast<ColumnFile&>(*destination);
+ DCHECK(output.empty());
+ output.reserve(count);
+ // Row copies preserve selection order, duplicates and the binary child's
owned bytes.
+ for (size_t i = 0; i < count; ++i) {
+ output.insert_from(*this, selector[i]);
+ }
+ return Status::OK();
+}
+
+MutableColumnPtr ColumnFile::permute(const Permutation& permutation, size_t
limit) const {
+ MutableColumns children;
+ children.reserve(NUM_CHILDREN);
+ for (const auto& column : _columns) {
+ children.push_back(column->permute(permutation, limit));
+ }
+ return create(std::move(children));
+}
+
+ColumnPtr ColumnFile::convert_column_if_overflow() {
+ for (auto& column : _columns) {
+ column = column->convert_column_if_overflow();
+ }
+ return IColumn::convert_column_if_overflow();
+}
+
+size_t ColumnFile::byte_size() const {
+ size_t size = 0;
+ for (const auto& column : _columns) {
+ size += column->byte_size();
+ }
+ return size;
+}
+
+size_t ColumnFile::allocated_bytes() const {
+ size_t size = 0;
+ for (const auto& column : _columns) {
+ size += column->allocated_bytes();
+ }
+ return size;
+}
+
+bool ColumnFile::has_enough_capacity(const IColumn& src) const {
+ const auto& file = assert_cast<const ColumnFile&>(src);
+ for (size_t i = 0; i < NUM_CHILDREN; ++i) {
+ if (!_columns[i]->has_enough_capacity(file.get_column(i))) {
+ return false;
+ }
+ }
+ return true;
+}
+
+void ColumnFile::for_each_subcolumn(ColumnCallback callback) {
+ for (auto& column : _columns) {
+ callback(column);
+ }
+}
+
+StringRef ColumnFile::serialize_value_into_arena(size_t n, Arena& arena, const
char*& begin) const {
+ char* pos = arena.alloc_continue(serialize_size_at(n), begin);
+ return {pos, serialize_impl(pos, n)};
+}
+
+const char* ColumnFile::deserialize_and_insert_from_arena(const char* pos) {
+ return pos + deserialize_impl(pos);
+}
+
+size_t ColumnFile::serialize_size_at(size_t row) const {
+ size_t size = 0;
+ for (const auto& column : _columns) {
+ size += column->serialize_size_at(row);
+ }
+ return size;
+}
+
+size_t ColumnFile::serialize_impl(char* pos, size_t row) const {
+ size_t size = 0;
+ for (const auto& column : _columns) {
+ size += column->serialize_impl(pos + size, row);
+ }
+ DCHECK_EQ(size, serialize_size_at(row));
+ return size;
+}
+
+size_t ColumnFile::deserialize_impl(const char* pos) {
+ size_t size = 0;
+ for (auto& column : _columns) {
+ size += column->deserialize_impl(pos + size);
+ }
+ return size;
+}
+
+void ColumnFile::serialize_vec(StringRef* keys, size_t num_rows) const {
+ for (size_t i = 0; i < num_rows; ++i) {
+ // The caller owns these mutable row buffers, as in other column
serializers.
+ keys[i].size += serialize_impl(const_cast<char*>(keys[i].data +
keys[i].size), i);
+ }
+}
+
+void ColumnFile::deserialize_vec(StringRef* keys, size_t num_rows) {
+ for (size_t i = 0; i < num_rows; ++i) {
+ const auto size = deserialize_impl(keys[i].data);
+ keys[i].data += size;
+ keys[i].size -= size;
+ }
+}
+
+size_t ColumnFile::get_max_row_byte_size() const {
+ size_t size = 0;
+ for (const auto& column : _columns) {
+ size += column->get_max_row_byte_size();
+ }
+ return size;
+}
+
+void ColumnFile::update_hash_with_value(size_t n, SipHash& hash) const {
+ hash_file_rows(*this, n, n + 1, nullptr,
+ [&](size_t, const char* data, size_t size) {
hash.update(data, size); });
+}
+
+void ColumnFile::update_hashes_with_value(uint64_t* __restrict hashes,
+ const uint8_t* __restrict null_data)
const {
+ hash_file_rows(*this, 0, size(), null_data, [&](size_t row, const char*
data, size_t size) {
+ hashes[row] = HashUtil::xxHash64WithSeed(data, size, hashes[row]);
+ });
+}
+
+void ColumnFile::update_xxHash_with_value(size_t start, size_t end, uint64_t&
hash,
Review Comment:
[P2] Fence direct distinct aggregates from FILE integrity hashes
The new physical hash is also reachable through this directly callable SQL
aggregate:
```text
Aggregate(output=[MULTI_DISTINCT_COUNT(ARRAY(f))])
Project(CAST(NAMED_STRUCT('uri',CONCAT('urn:',number),
'offset',NULL,'size',NULL,'content_type',NULL,
'checksum',NULL,'inline',NULL) AS FILE) AS f)
Numbers(number=2)
```
`MultiDistinctCount` is registered as a builtin, accepts ANY, forces its
distinct flag to false, and checks only legacy VARIANT. BE's uniq factory
supports TYPE_ARRAY and calls `ColumnArray::update_xxHash_with_value` -> this
method, so FILE's complete serialized bytes become a SQL deduplication key
despite the declared integrity-only hash contract. Recursively reject FILE in
this direct aggregate's own legality check. This is distinct from the existing
ARRAY_AGG(DISTINCT f) report: there is no DISTINCT modifier or
optimizer-created grouping key, and checking `getDistinctArguments()` would
leave this instance admitted.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/runtime/ThriftPlansBuilder.java:
##########
@@ -239,6 +240,10 @@ static Map<String, TAIResource>
collectAiResources(ConnectContext connectContext
return aiResourceMap;
}
+ public static Map<String, TS3Resource> collectS3Resources(ConnectContext
context) {
+ return context == null ? Maps.newLinkedHashMap() :
context.getS3Resources();
Review Comment:
[P2] Preserve job-owned resources across Broker Load workers
A Broker LOAD mapping such as `SET(sz=ELEMENT_AT(TO_FILE('r',uri),'size'))`
targets BIGINT and passes the FILE storage/output restrictions. Its mapping is
translated by `NereidsLoadingTaskPlanner`/`NereidsLoadPlanInfoCollector` on the
pending-task worker, but the resource is registered only in that worker's
ConnectContext. `LoadLoadingTask.executeOnce` later runs on a loading worker:
the default NereidsCoordinator captures that worker's context through
`CoordinatorContext.buildForLoad`, so this helper sends its empty resource map;
the legacy Coordinator constructor instead leaves context null. Neither
transfers the planning snapshot, and TO_FILE cannot resolve `r` in BE. A fresh
planning worker also lacks the job's currentUserIdentity
(`BrokerLoadJob.setComputeGroup` restores only compute selection), so the new
USAGE check can fail earlier on a null identity. Restore the submitting
identity for planning and carry an immutable task-owned snapshot into
execution, with a Broker LOAD
mapping test. This first-run asynchronous Broker Load boundary is separate
from the existing routine-load refresh and external opaque-plan reports.
--
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]
