Skip to content

Commit c43019e

Browse files
yhmoyhmo
committed
Support AddFunctionField/DropCollectionField/DropFunctionField
Signed-off-by: yhmo <yihua.mo@zilliz.com>
1 parent ee4b440 commit c43019e

10 files changed

Lines changed: 727 additions & 8 deletions

File tree

‎examples/src/main/java/io/milvus/v2/CDCExample.java‎

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,19 +22,25 @@
2222
import io.milvus.v2.client.ConnectConfig;
2323
import io.milvus.v2.client.MilvusClientV2;
2424
import io.milvus.v2.service.cdc.request.CrossClusterTopology;
25+
import io.milvus.v2.service.cdc.request.DumpMessagesReq;
26+
import io.milvus.v2.service.cdc.request.GetReplicateInfoReq;
2527
import io.milvus.v2.service.cdc.request.MilvusCluster;
2628
import io.milvus.v2.service.cdc.request.ReplicateConfiguration;
2729
import io.milvus.v2.service.cdc.request.UpdateReplicateConfigurationReq;
30+
import io.milvus.v2.service.cdc.response.DumpMessageInfo;
31+
import io.milvus.v2.service.cdc.response.DumpMessagesResp;
32+
import io.milvus.v2.service.cdc.response.GetReplicateConfigurationResp;
33+
import io.milvus.v2.service.cdc.response.GetReplicateInfoResp;
2834

2935
import java.util.ArrayList;
3036
import java.util.List;
3137

3238
public class CDCExample {
33-
private static final String clusterAURI = "http://192.168.1.1:19530";
34-
private static final String clusterBURI = "http://192.168.1.1:19500";
39+
private static final String clusterAURI = "http://localhost:19530";
40+
private static final String clusterBURI = "http://localhost:29530";
3541

36-
private static final String clusterAId = "cdc-test-upstream";
37-
private static final String clusterBId = "cdc-test-downstream";
42+
private static final String clusterAId = "cdc-a";
43+
private static final String clusterBId = "cdc-b";
3844

3945
private static final Integer pchannelNum = 16;
4046

@@ -51,11 +57,13 @@ public static void main(String[] args) {
5157
.uri(clusterAURI)
5258
.build();
5359
MilvusClientV2 clusterAClient = new MilvusClientV2(clusterA);
60+
System.out.println("Cluster A connected: " + clusterAClient.getServerVersion());
5461

5562
ConnectConfig clusterB = ConnectConfig.builder()
5663
.uri(clusterBURI)
5764
.build();
5865
MilvusClientV2 clusterBClient = new MilvusClientV2(clusterB);
66+
System.out.println("Cluster B connected: " + clusterBClient.getServerVersion());
5967

6068
MilvusCluster milvusClusterA = MilvusCluster.builder()
6169
.clusterId(clusterAId)
@@ -88,6 +96,32 @@ public static void main(String[] args) {
8896
.build();
8997

9098
clusterAClient.updateReplicateConfiguration(updateReq);
99+
System.out.println("Replicate configuration updated for cluster A");
91100
clusterBClient.updateReplicateConfiguration(updateReq);
101+
System.out.println("Replicate configuration updated for cluster B");
102+
//
103+
GetReplicateConfigurationResp replicateConfigurationResp = clusterAClient.getReplicateConfiguration();
104+
System.out.println("Replicate configuration: " + replicateConfigurationResp.getReplicateConfiguration());
105+
106+
GetReplicateInfoResp replicateInfoResp = clusterBClient.getReplicateInfo(GetReplicateInfoReq.builder()
107+
.sourceClusterId(clusterAId)
108+
.targetPchannel(generatePChannels(clusterBId).get(0))
109+
.build());
110+
System.out.println("Replicate info: " + replicateInfoResp);
111+
112+
String dumpStartMessageId = replicateInfoResp.getSalvageCheckpoint().getMessageID().getId();
113+
114+
DumpMessagesResp dumpMessagesResp = clusterAClient.dumpMessages(DumpMessagesReq.builder()
115+
.pchannel(generatePChannels(clusterAId).get(0))
116+
.startMessageID(GetReplicateInfoResp.MessageID.builder()
117+
.id(dumpStartMessageId)
118+
.walName("RocksMQ")
119+
.build())
120+
.build());
121+
System.out.println("Dump messages:");
122+
for (DumpMessageInfo message : dumpMessagesResp) {
123+
System.out.println("\tmessage id: " + message.getMessageID().getId());
124+
System.out.println("\tpayload size: " + message.getPayload().length);
125+
}
92126
}
93127
}

‎sdk-core/src/main/java/io/milvus/v2/client/MilvusClientV2.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -509,6 +509,15 @@ public void alterCollectionField(AlterCollectionFieldReq request) {
509509
rpcUtils.retry(() -> collectionService.alterCollectionField(this.getRpcStub(), request));
510510
}
511511

512+
/**
513+
* Drop a field from collection.
514+
*
515+
* @param request drop field request
516+
*/
517+
public void dropCollectionField(DropCollectionFieldReq request) {
518+
rpcUtils.retry(() -> collectionService.dropCollectionField(this.getRpcStub(), request));
519+
}
520+
512521
/**
513522
* drop a collection's properties.
514523
*
@@ -643,6 +652,15 @@ public void addCollectionFunction(AddCollectionFunctionReq request) {
643652
rpcUtils.retry(() -> collectionService.addCollectionFunction(this.getRpcStub(), request));
644653
}
645654

655+
/**
656+
* Add a function-backed field to collection.
657+
*
658+
* @param request add function field request
659+
*/
660+
public void addFunctionField(AddFunctionFieldReq request) {
661+
rpcUtils.retry(() -> collectionService.addFunctionField(this.getRpcStub(), request));
662+
}
663+
646664
/**
647665
* Alter a function of collection.
648666
*
@@ -654,13 +672,25 @@ public void alterCollectionFunction(AlterCollectionFunctionReq request) {
654672

655673
/**
656674
* Drop a function of collection.
675+
* This removes only the function definition and keeps its output field(s).
657676
*
658677
* @param request drop function request
659678
*/
660679
public void dropCollectionFunction(DropCollectionFunctionReq request) {
661680
rpcUtils.retry(() -> collectionService.dropCollectionFunction(this.getRpcStub(), request));
662681
}
663682

683+
/**
684+
* Drop a function-backed field from collection.
685+
* This removes the function and its output field(s), unlike dropCollectionFunction,
686+
* which only drops the function definition.
687+
*
688+
* @param request drop function field request
689+
*/
690+
public void dropFunctionField(DropFunctionFieldReq request) {
691+
rpcUtils.retry(() -> collectionService.dropFunctionField(this.getRpcStub(), request));
692+
}
693+
664694
/////////////////////////////////////////////////////////////////////////////////////////////
665695
// Index Operations
666696
/////////////////////////////////////////////////////////////////////////////////////////////

‎sdk-core/src/main/java/io/milvus/v2/service/collection/CollectionService.java‎

Lines changed: 113 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -343,6 +343,40 @@ public Void alterCollectionField(MilvusServiceGrpc.MilvusServiceBlockingStub blo
343343
return null;
344344
}
345345

346+
public Void dropCollectionField(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub, DropCollectionFieldReq request) {
347+
String dbName = request.getDatabaseName();
348+
String collectionName = request.getCollectionName();
349+
String fieldName = request.getFieldName();
350+
Long fieldId = request.getFieldId();
351+
boolean hasFieldName = StringUtils.isNotBlank(fieldName);
352+
boolean hasFieldId = fieldId != null && fieldId > 0;
353+
if (hasFieldName == hasFieldId) {
354+
throw new MilvusClientException(ErrorCode.INVALID_PARAMS,
355+
"Exactly one of fieldName or fieldId must be provided.");
356+
}
357+
358+
String title = String.format("Drop field of collection: '%s' in database: '%s'", collectionName, dbName);
359+
AlterCollectionSchemaRequest.DropRequest.Builder dropBuilder = AlterCollectionSchemaRequest.DropRequest.newBuilder();
360+
if (hasFieldName) {
361+
dropBuilder.setFieldName(fieldName);
362+
} else {
363+
dropBuilder.setFieldId(fieldId);
364+
}
365+
366+
AlterCollectionSchemaRequest.Builder builder = AlterCollectionSchemaRequest.newBuilder()
367+
.setCollectionName(collectionName)
368+
.setAction(AlterCollectionSchemaRequest.Action.newBuilder()
369+
.setDropRequest(dropBuilder)
370+
.build());
371+
if (StringUtils.isNotEmpty(dbName)) {
372+
builder.setDbName(dbName);
373+
}
374+
375+
AlterCollectionSchemaResponse response = blockingStub.alterCollectionSchema(builder.build());
376+
rpcUtils.handleResponse(title, response.getAlterStatus());
377+
return null;
378+
}
379+
346380
public Void dropCollectionProperties(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub, DropCollectionPropertiesReq request) {
347381
String dbName = request.getDatabaseName();
348382
String collectionName = request.getCollectionName();
@@ -697,6 +731,49 @@ public Void addCollectionFunction(MilvusServiceGrpc.MilvusServiceBlockingStub bl
697731
return null;
698732
}
699733

734+
public Void addFunctionField(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub,
735+
AddFunctionFieldReq request) {
736+
if (request.getFunction() == null) {
737+
throw new MilvusClientException(ErrorCode.INVALID_PARAMS, "Function cannot be null.");
738+
}
739+
if (CollectionUtils.isEmpty(request.getFunction().getOutputFieldNames())
740+
|| request.getFunction().getOutputFieldNames().size() != 1
741+
|| !StringUtils.equals(request.getFieldName(), request.getFunction().getOutputFieldNames().get(0))) {
742+
throw new MilvusClientException(ErrorCode.INVALID_PARAMS,
743+
"Function must have exactly one output field matching the field being added.");
744+
}
745+
746+
String dbName = request.getDatabaseName();
747+
String collectionName = request.getCollectionName();
748+
String title = String.format("Add function field to collection: '%s' in database: '%s'", collectionName, dbName);
749+
750+
CreateCollectionReq.FieldSchema fieldSchema = SchemaUtils.convertFieldReqToFieldSchema(request);
751+
FieldSchema grpcFieldSchema = SchemaUtils.convertToGrpcFieldSchema(fieldSchema)
752+
.toBuilder()
753+
.setIsFunctionOutput(true)
754+
.build();
755+
AlterCollectionSchemaRequest.FieldInfo fieldInfo = AlterCollectionSchemaRequest.FieldInfo.newBuilder()
756+
.setFieldSchema(grpcFieldSchema)
757+
.build();
758+
AlterCollectionSchemaRequest.AddRequest addRequest = AlterCollectionSchemaRequest.AddRequest.newBuilder()
759+
.addFieldInfos(fieldInfo)
760+
.addFuncSchema(SchemaUtils.convertToGrpcFunction(request.getFunction()))
761+
.build();
762+
763+
AlterCollectionSchemaRequest.Builder builder = AlterCollectionSchemaRequest.newBuilder()
764+
.setCollectionName(collectionName)
765+
.setAction(AlterCollectionSchemaRequest.Action.newBuilder()
766+
.setAddRequest(addRequest)
767+
.build());
768+
if (StringUtils.isNotEmpty(dbName)) {
769+
builder.setDbName(dbName);
770+
}
771+
772+
AlterCollectionSchemaResponse response = blockingStub.alterCollectionSchema(builder.build());
773+
rpcUtils.handleResponse(title, response.getAlterStatus());
774+
return null;
775+
}
776+
700777
public Void alterCollectionFunction(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub,
701778
AlterCollectionFunctionReq request) {
702779
if (request.getFunction() == null) {
@@ -728,15 +805,47 @@ public Void dropCollectionFunction(MilvusServiceGrpc.MilvusServiceBlockingStub b
728805
String dbName = request.getDatabaseName();
729806
String collectionName = request.getCollectionName();
730807
String title = String.format("Drop function to collection: '%s' in database: '%s'", collectionName, dbName);
731-
DropCollectionFunctionRequest.Builder builder = DropCollectionFunctionRequest.newBuilder()
808+
AlterCollectionSchemaRequest.Builder builder = AlterCollectionSchemaRequest.newBuilder()
732809
.setCollectionName(collectionName)
733-
.setFunctionName(request.getFunctionName());
810+
.setAction(AlterCollectionSchemaRequest.Action.newBuilder()
811+
.setDropRequest(AlterCollectionSchemaRequest.DropRequest.newBuilder()
812+
.setFunctionName(request.getFunctionName())
813+
.setDropFunctionOutputFields(false)
814+
.build())
815+
.build());
816+
if (StringUtils.isNotEmpty(dbName)) {
817+
builder.setDbName(dbName);
818+
}
819+
820+
AlterCollectionSchemaResponse response = blockingStub.alterCollectionSchema(builder.build());
821+
rpcUtils.handleResponse(title, response.getAlterStatus());
822+
823+
return null;
824+
}
825+
826+
public Void dropFunctionField(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub,
827+
DropFunctionFieldReq request) {
828+
if (StringUtils.isEmpty(request.getFunctionName())) {
829+
throw new MilvusClientException(ErrorCode.INVALID_PARAMS, "Function name cannot be empty.");
830+
}
831+
832+
String dbName = request.getDatabaseName();
833+
String collectionName = request.getCollectionName();
834+
String title = String.format("Drop function field of collection: '%s' in database: '%s'", collectionName, dbName);
835+
AlterCollectionSchemaRequest.Builder builder = AlterCollectionSchemaRequest.newBuilder()
836+
.setCollectionName(collectionName)
837+
.setAction(AlterCollectionSchemaRequest.Action.newBuilder()
838+
.setDropRequest(AlterCollectionSchemaRequest.DropRequest.newBuilder()
839+
.setFunctionName(request.getFunctionName())
840+
.setDropFunctionOutputFields(true)
841+
.build())
842+
.build());
734843
if (StringUtils.isNotEmpty(dbName)) {
735844
builder.setDbName(dbName);
736845
}
737-
Status status = blockingStub.dropCollectionFunction(builder.build());
738-
rpcUtils.handleResponse(title, status);
739846

847+
AlterCollectionSchemaResponse response = blockingStub.alterCollectionSchema(builder.build());
848+
rpcUtils.handleResponse(title, response.getAlterStatus());
740849
return null;
741850
}
742851

‎sdk-core/src/main/java/io/milvus/v2/service/collection/request/AddFunctionFieldReq.java‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package io.milvus.v2.service.collection.request;
21+
22+
public class AddFunctionFieldReq extends AddFieldReq {
23+
private String collectionName;
24+
private String databaseName;
25+
private CreateCollectionReq.Function function;
26+
27+
private AddFunctionFieldReq(AddFunctionFieldReqBuilder builder) {
28+
super(builder);
29+
this.collectionName = builder.collectionName;
30+
this.databaseName = builder.databaseName;
31+
this.function = builder.function;
32+
}
33+
34+
public String getCollectionName() {
35+
return collectionName;
36+
}
37+
38+
public void setCollectionName(String collectionName) {
39+
this.collectionName = collectionName;
40+
}
41+
42+
public String getDatabaseName() {
43+
return databaseName;
44+
}
45+
46+
public void setDatabaseName(String databaseName) {
47+
this.databaseName = databaseName;
48+
}
49+
50+
public CreateCollectionReq.Function getFunction() {
51+
return function;
52+
}
53+
54+
public void setFunction(CreateCollectionReq.Function function) {
55+
this.function = function;
56+
}
57+
58+
@Override
59+
public String toString() {
60+
return "AddFunctionFieldReq{" +
61+
"collectionName='" + collectionName + '\'' +
62+
", databaseName='" + databaseName + '\'' +
63+
", function=" + function +
64+
", " + super.toString() +
65+
'}';
66+
}
67+
68+
public static AddFunctionFieldReqBuilder builder() {
69+
return new AddFunctionFieldReqBuilder();
70+
}
71+
72+
public static class AddFunctionFieldReqBuilder extends AddFieldReq.AddFieldReqBuilder<AddFunctionFieldReqBuilder> {
73+
private String collectionName = "";
74+
private String databaseName = "";
75+
private CreateCollectionReq.Function function;
76+
77+
private AddFunctionFieldReqBuilder() {
78+
}
79+
80+
public AddFunctionFieldReqBuilder collectionName(String collectionName) {
81+
this.collectionName = collectionName;
82+
return this;
83+
}
84+
85+
public AddFunctionFieldReqBuilder databaseName(String databaseName) {
86+
this.databaseName = databaseName;
87+
return this;
88+
}
89+
90+
public AddFunctionFieldReqBuilder function(CreateCollectionReq.Function function) {
91+
this.function = function;
92+
return this;
93+
}
94+
95+
@Override
96+
public AddFunctionFieldReq build() {
97+
return new AddFunctionFieldReq(this);
98+
}
99+
}
100+
}

0 commit comments

Comments
 (0)