github-actions[bot] commented on code in PR #67673:
URL: https://github.com/apache/doris/pull/67673#discussion_r3978208049
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/AIResource.java:
##########
@@ -127,7 +135,8 @@ public void modifyProperties(Map<String, String>
properties) throws DdlException
writeLock();
for (Map.Entry<String, String> kv : properties.entrySet()) {
replaceIfEffectiveValue(this.properties, kv.getKey(),
kv.getValue());
- if (kv.getKey().equals(AIProperties.API_KEY)) {
+ if (kv.getKey().equals(AIProperties.API_KEY)
+ || kv.getKey().equals(AIProperties.EMBED_API_KEY)) {
Review Comment:
[P2] Validate the merged resource under the write lock. Two concurrent,
individually valid ALTERs can currently commit an invalid embed group: from
OPENAI plus a key, one request can set LOCAL and clear the key while another
sets GEMINI; both validate the old snapshot, then LOCAL/empty followed by
GEMINI leaves a remote provider with no key. Both calls succeed and the invalid
resource is journaled. Build, validate, normalize, and install the merged
snapshot as one locked transition, and add an interleaving test.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/AIProperties.java:
##########
@@ -53,29 +58,31 @@ public class AIProperties extends BaseProperties {
public static final String VALIDITY_CHECK = "ai.validity_check";
public static final List<String> REQUIRED_FIELDS = Arrays.asList(ENDPOINT,
PROVIDER_TYPE, MODEL_NAME);
+ public static final List<String> EMBED_REQUIRED_FIELDS =
+ Arrays.asList(EMBED_ENDPOINT, EMBED_PROVIDER_TYPE,
EMBED_MODEL_NAME);
public static final List<String> PROVIDERS
= Arrays.asList("OPENAI", "LOCAL", "GEMINI", "DEEPSEEK",
"ANTHROPIC",
"MOONSHOT", "QWEN", "MINIMAX", "ZHIPU", "BAICHUAN", "VOYAGEAI",
"JINA");
+ public static final List<String> EFFORT_LEVELS =
+ Arrays.asList("none", "minimal", "low", "medium", "high", "xhigh",
"max");
public static void requiredAIProperties(Map<String, String> properties)
throws DdlException {
- // Check required field
- for (String field : REQUIRED_FIELDS) {
- if (Strings.isNullOrEmpty(properties.get(field))) {
- throw new DdlException("Missing [" + field + "] in
properties.");
- }
+ boolean hasGeneralProperties = hasAnyProperty(properties,
REQUIRED_FIELDS, API_KEY);
+ boolean hasEmbedProperties = hasAnyProperty(properties,
EMBED_REQUIRED_FIELDS, EMBED_API_KEY);
+ if (!hasGeneralProperties && !hasEmbedProperties) {
+ throw new DdlException("At least one complete AI property group
must be configured.");
}
- // Check the provider is valid
- properties.put(PROVIDER_TYPE,
properties.get(PROVIDER_TYPE).toUpperCase());
- if (PROVIDERS.stream().noneMatch(s ->
s.equals(properties.get(PROVIDER_TYPE).toUpperCase()))) {
- throw new DdlException("Provider must be one of " + PROVIDERS);
+ if (hasGeneralProperties) {
+ validatePropertyGroup(properties, REQUIRED_FIELDS, PROVIDER_TYPE,
API_KEY);
+ }
+ if (hasEmbedProperties) {
+ validatePropertyGroup(properties, EMBED_REQUIRED_FIELDS,
EMBED_PROVIDER_TYPE, EMBED_API_KEY);
Review Comment:
[P2] Validate dedicated groups against embedding-capable providers. This
path accepts every generation provider, including DEEPSEEK and MOONSHOT, but
their BE adapters explicitly return NotSupported from both embedding methods.
An embed-only resource therefore passes CREATE and analysis yet can never
execute. Use a capability-specific allowlist (or shared capability metadata)
and cover rejected providers in FE tests.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/ai/AIFunction.java:
##########
@@ -61,6 +61,10 @@ public void checkLegalityAfterRewrite() {
if (!(resource instanceof AIResource)) {
throw new AnalysisException("AI resource '" + resourceName +
"' does not exist");
}
+ if (!((AIResource) resource).hasCompleteGeneralProperties()) {
Review Comment:
[P1] Pin the resource validated for this statement. The check is against one
object, but only its name is registered; a concurrent DROP followed by CREATE
of a valid embed-only resource under the same name makes both transport
builders serialize the replacement. Scalar AI then creates an adapter from an
empty general provider and hits `DORIS_CHECK(adapter)` (`AI_AGG` dereferences
it). Store the validated snapshot/identity in `StatementContext`, or revalidate
and snapshot atomically at transport, and add a DROP/CREATE interleaving test.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/AIResource.java:
##########
@@ -159,10 +169,33 @@ protected void getProcNodeData(BaseProcResult result) {
public TAIResource toThrift() throws NumberFormatException {
TAIResource tAIResource = new TAIResource();
-
tAIResource.setProviderType(properties.get(AIProperties.PROVIDER_TYPE));
- tAIResource.setEndpoint(properties.get(AIProperties.ENDPOINT));
- tAIResource.setApiKey(properties.get(AIProperties.API_KEY));
- tAIResource.setModelName(properties.get(AIProperties.MODEL_NAME));
+ if (properties.containsKey(AIProperties.PROVIDER_TYPE)) {
+
tAIResource.setProviderType(properties.get(AIProperties.PROVIDER_TYPE));
+ }
+ if (properties.containsKey(AIProperties.ENDPOINT)) {
+ tAIResource.setEndpoint(properties.get(AIProperties.ENDPOINT));
+ }
+ if (properties.containsKey(AIProperties.API_KEY)) {
+ tAIResource.setApiKey(properties.get(AIProperties.API_KEY));
+ }
+ if (properties.containsKey(AIProperties.MODEL_NAME)) {
+ tAIResource.setModelName(properties.get(AIProperties.MODEL_NAME));
+ }
+ if (properties.containsKey(AIProperties.EMBED_PROVIDER_TYPE)) {
+
tAIResource.setEmbedProviderType(properties.get(AIProperties.EMBED_PROVIDER_TYPE));
+ }
+ if (properties.containsKey(AIProperties.EMBED_ENDPOINT)) {
+
tAIResource.setEmbedEndpoint(properties.get(AIProperties.EMBED_ENDPOINT));
+ }
+ if (properties.containsKey(AIProperties.EMBED_API_KEY)) {
+
tAIResource.setEmbedApiKey(properties.get(AIProperties.EMBED_API_KEY));
+ }
+ if (properties.containsKey(AIProperties.EMBED_MODEL_NAME)) {
+
tAIResource.setEmbedModelName(properties.get(AIProperties.EMBED_MODEL_NAME));
+ }
+ if (!Strings.isNullOrEmpty(properties.get(AIProperties.EFFORT))) {
Review Comment:
[P2] Preserve an empty effort as an actual clear on ALTER. The validator
accepts `ai.effort=""` and this branch defines empty as omitted, but
`modifyProperties` routes it through `replaceIfEffectiveValue`, which ignores
empty strings; only the API-key fields get an explicit empty-value override.
After setting `high`, an ALTER to `""` succeeds and increments the version
while `high` remains stored and forwarded. Remove/store this optional field
explicitly and test set -> empty through persistence and
`toThrift().isSetEffort()`.
##########
be/test/ai/aggregate_function_ai_agg_test.cpp:
##########
@@ -214,6 +214,136 @@ TEST_F(AggregateFunctionAIAggTest,
serialize_deserialize_test) {
_agg_function->destroy(place2);
}
+TEST_F(AggregateFunctionAIAggTest,
serialize_effort_only_for_supported_exec_version) {
+ TAIResource ai_resource;
+ ai_resource.provider_type = "MOCK";
+ ai_resource.model_name = "mock_model";
+ ai_resource.endpoint = "http://localhost";;
+ ai_resource.api_key = "xxx";
+ ai_resource.temperature = 0.5;
+ ai_resource.max_tokens = 16;
+ ai_resource.max_retries = 1;
+ ai_resource.retry_delay_second = 1;
+ ai_resource.dimensions = 514;
+ ai_resource.effort = "high";
+ _query_ctx->set_ai_resources(
+ std::map<std::string, TAIResource> {{"effort_resource",
ai_resource}});
+
+ auto resource_col = ColumnString::create();
+ auto text_col = ColumnString::create();
+ auto task_col = ColumnString::create();
+ resource_col->insert_data("effort_resource", 15);
+ text_col->insert_data("test", 4);
+ task_col->insert_data("summarize", 9);
+
+ std::unique_ptr<char[]> memory(new char[_agg_function->size_of_data()]);
+ AggregateDataPtr place = memory.get();
+ _agg_function->create(place);
+ const IColumn* columns[3] = {resource_col.get(), text_col.get(),
task_col.get()};
+ _agg_function->add(place, columns, 0, _arena);
+
+ _agg_function->set_version(14);
+ auto legacy_column = _agg_function->create_serialize_column();
+ _agg_function->serialize_without_key_to_column(place, *legacy_column);
+
+ _agg_function->set_version(15);
+ auto effort_column = _agg_function->create_serialize_column();
+ _agg_function->serialize_without_key_to_column(place, *effort_column);
+
+ const StringRef legacy_state = legacy_column->get_data_at(0);
+ const StringRef effort_state = effort_column->get_data_at(0);
+
+ ColumnString serialized_effort;
+ VectorBufferWriter effort_writer(serialized_effort);
+ effort_writer.write_binary(std::string("high"));
+ effort_writer.commit();
+ const StringRef expected_suffix = serialized_effort.get_data_at(0);
+
+ ASSERT_EQ(effort_state.size, legacy_state.size + expected_suffix.size);
+ EXPECT_EQ(std::string_view(effort_state.data + legacy_state.size,
expected_suffix.size),
+ std::string_view(expected_suffix.data, expected_suffix.size));
+
+ _agg_function->destroy(place);
+}
+
+TEST_F(AggregateFunctionAIAggTest,
deserialize_legacy_multi_row_state_does_not_read_effort) {
+ TAIResource ai_resource;
+ ai_resource.provider_type = "MOCK";
+ ai_resource.model_name = "mock_model";
+ ai_resource.endpoint = "http://localhost";;
+ ai_resource.api_key = "xxx";
+ ai_resource.temperature = 0.5;
+ ai_resource.max_tokens = 16;
+ ai_resource.max_retries = 1;
+ ai_resource.retry_delay_second = 1;
+ ai_resource.dimensions = 514;
+ ai_resource.effort = "high";
+ _query_ctx->set_ai_resources(
+ std::map<std::string, TAIResource> {{"effort_resource",
ai_resource}});
+
+ auto resource_col = ColumnString::create();
+ auto text_col = ColumnString::create();
+ auto task_col = ColumnString::create();
+ resource_col->insert_data("effort_resource", 15);
+ text_col->insert_data("test", 4);
+ task_col->insert_data("summarize", 9);
+
+ std::unique_ptr<char[]> source_memory(new
char[_agg_function->size_of_data()]);
+ AggregateDataPtr source_place = source_memory.get();
+ _agg_function->create(source_place);
+ const IColumn* columns[3] = {resource_col.get(), text_col.get(),
task_col.get()};
+ _agg_function->add(source_place, columns, 0, _arena);
+
+ _agg_function->set_version(15);
+ auto current_format_column = _agg_function->create_serialize_column();
+ _agg_function->serialize_without_key_to_column(source_place,
*current_format_column);
+ const StringRef current_format_state =
current_format_column->get_data_at(0);
+
+ ColumnString serialized_effort;
+ VectorBufferWriter effort_writer(serialized_effort);
+ effort_writer.write_binary(std::string("high"));
+ effort_writer.commit();
+ const size_t effort_size = serialized_effort.get_data_at(0).size;
+
+ auto legacy_column = ColumnString::create();
+ for (int i = 0; i < 3; ++i) {
+ legacy_column->insert_data(current_format_state.data,
+ current_format_state.size - effort_size);
+ }
+
+ std::unique_ptr<char[]> merged_memory(new
char[_agg_function->size_of_data()]);
+ AggregateDataPtr merged_place = merged_memory.get();
+ _agg_function->create(merged_place);
+ _agg_function->set_version(14);
+ _agg_function->deserialize_and_merge_from_column_range(merged_place,
*legacy_column, 0, 1,
+ _arena);
+
+ const auto& merged_data =
Review Comment:
[P2] Apply clang-format v16 here. The exact-head Clang Formatter check is
failing, and a local non-mutating run over all changed BE files isolates this
declaration; the repository formatter places the declaration and cast on one
line.
--
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]
