Skip to content

SK-2302 Multi insert support in v3 #236

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

Closed
skyflow-bharti wants to merge 4 commits into from
Closed

SK-2302 Multi insert support in v3 #236

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
ae75c8d
SK-2302 add changes for bulk async changes
skyflow-bharti Sep 19, 2025
e561622
SK-2302 updated the messages
skyflow-bharti Sep 19, 2025
a5514d3
SK-2302 updated the messages
skyflow-bharti Sep 19, 2025
c5595d0
SK-2302 updated variable names
skyflow-bharti Sep 19, 2025
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
8 changes: 6 additions & 2 deletions common/src/main/java/com/skyflow/errors/ErrorMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,8 @@ public enum ErrorMessage {
ConnectionIdNotInConfigList("%s0 Validation error. ConnectionId is missing from the config. Specify the connectionIds from configs."),
EmptyCredentials("%s0 Validation error. Invalid credentials. Credentials must not be empty."),
TableSpecifiedInRequestAndRecordObject("%s0 Validation error. Table name cannot be specified at both the request and record levels. Please specify the table name in only one place."),
UpsertAtRecordLevel("%s0 Validation error. Upsert specify "),
UpsertTableRequestAtRecordLevel("%s0 Validation error. Table name should be present at each record level when upsert is present at record level."),
UpsertTableRequestAtRequestLevel("%S0 Validation error. Upsert should be present at each record level when table name is present at record level."),
UpsertTableRequestAtRequestLevel("%s0 Validation error. Upsert should be present at each record level when table name is present at record level."),
TableNotSpecifiedInRequestAndRecordObject("%s0 Validation error. Table name is missing. Table name should be specified at one place either at the request level or record level. Please specify the table name at one place."),
// Vault config
InvalidVaultId("%s0 Initialization failed. Invalid vault ID. Specify a valid vault ID."),
Expand Down Expand Up @@ -64,6 +63,10 @@ public enum ErrorMessage {
TableKeyError("%s0 Validation error. 'table' key is missing from the payload. Specify a 'table' key."),
EmptyTable("%s0 Validation error. 'table' can't be empty. Specify a table."),
ValuesKeyError("%s0 Validation error. 'values' key is missing from the payload. Specify a 'values' key."),
EmptyRecords("%s0 Validation error. 'records' can't be empty. Specify records."),
EmptyKeyInRecords("%s0 Validation error. Invalid key in data in records. Specify a valid key."),
EmptyValueInRecords("%s0 Validation error. Invalid value in records. Specify a valid value."),
RecordsKeyError("%s0 Validation error. 'records' key is missing from the payload. Specify a 'records' key."),
EmptyValues("%s0 Validation error. 'values' can't be empty. Specify values."),
EmptyKeyInValues("%s0 Validation error. Invalid key in values. Specify a valid key."),
EmptyValueInValues("%s0 Validation error. Invalid value in values. Specify a valid value."),
Expand Down Expand Up @@ -168,6 +171,7 @@ public enum ErrorMessage {
NullRedactionInTokenGroup("%s0 Validation error. Redaction in TokenGroupRedactions is null or empty. Specify a valid redaction."),

NullTokenGroupNameInTokenGroup("%s0 Validation error. TokenGroupName in TokenGroupRedactions is null or empty. Specify a valid tokenGroupName."),
InvalidRecord("%s0 Validation error. InsertRecord object in the list is invalid. Specify a valid InsertRecord object."),
;
;
private final String message;
Expand Down
3 changes: 3 additions & 0 deletions common/src/main/java/com/skyflow/logs/ErrorLogs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,9 @@ public enum ErrorLogs {
EMPTY_TABLE_NAME("Invalid %s1 request. Table name can not be empty."),
VALUES_IS_REQUIRED("Invalid %s1 request. Values are required."),
EMPTY_VALUES("Invalid %s1 request. Values can not be empty."),
RECORDS_IS_REQUIRED("Invalid %s1 request. Records are required."),
EMPTY_RECORDS("Invalid %s1 request. Records can not be empty."),
INVALID_RECORD("Invalid %s1 request. Invalid record. Specify a valid record."),
RECORD_SIZE_EXCEED("Maximum number of records exceeded. The limit is 10000."),
TOKENS_SIZE_EXCEED("Maximum number of tokens exceeded. The limit is 10000."),
EMPTY_OR_NULL_VALUE_IN_VALUES("Invalid %s1 request. Value can not be null or empty in values for key \"%s2\"."),
Expand Down
28 changes: 14 additions & 14 deletions v3/src/main/java/com/skyflow/VaultClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
package com.skyflow;

import java.util.ArrayList;
import java.util.List;

import com.skyflow.config.Credentials;
import com.skyflow.config.VaultConfig;
import com.skyflow.enums.UpdateType;
Expand Down Expand Up @@ -27,9 +30,6 @@
import okhttp3.OkHttpClient;
import okhttp3.Request;

import java.util.ArrayList;
import java.util.List;


public class VaultClient {
private final VaultConfig vaultConfig;
Expand Down Expand Up @@ -134,26 +134,26 @@ protected void updateExecutorInHTTP() {
}

protected InsertRequest getBulkInsertRequestBody(com.skyflow.vault.data.InsertRequest request, VaultConfig config) {
ArrayList<InsertRecord> values = request.getRecords();
ArrayList<InsertRecord> records = request.getRecords();
List<InsertRecordData> insertRecordDataList = new ArrayList<>();
for (InsertRecord value : values) {
for (InsertRecord record : records) {
InsertRecordData.Builder data = InsertRecordData.builder();
data.data(value.getData());
if (value.getTable() != null && !value.getTable().isEmpty()){
data.tableName(value.getTable());
data.data(record.getData());
if (record.getTable() != null && !record.getTable().isEmpty()){
data.tableName(record.getTable());
}
if (value.getUpsert() != null && !value.getUpsert().isEmpty()){
if (value.getUpsertType() != null) {
if (record.getUpsert() != null && !record.getUpsert().isEmpty()){
if (record.getUpsertType() != null) {
EnumUpdateType updateType = null;
if (request.getUpsertType() == UpdateType.REPLACE) {
if (record.getUpsertType() == UpdateType.REPLACE) {
updateType = EnumUpdateType.REPLACE;
} else if (request.getUpsertType() == UpdateType.UPDATE) {
} else if (record.getUpsertType() == UpdateType.UPDATE) {
updateType = EnumUpdateType.UPDATE;
}
Upsert upsert = Upsert.builder().uniqueColumns(value.getUpsert()).updateType(updateType).build();
Upsert upsert = Upsert.builder().uniqueColumns(record.getUpsert()).updateType(updateType).build();
data.upsert(upsert);
} else {
Upsert upsert = Upsert.builder().uniqueColumns(value.getUpsert()).build();
Upsert upsert = Upsert.builder().uniqueColumns(record.getUpsert()).build();
data.upsert(upsert);
}
}
Expand Down
2 changes: 0 additions & 2 deletions v3/src/main/java/com/skyflow/enums/UpdateType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
package com.skyflow.enums;

import com.fasterxml.jackson.annotation.JsonValue;

public enum UpdateType {
UPDATE("UPDATE"),

Expand Down
1 change: 0 additions & 1 deletion v3/src/main/java/com/skyflow/utils/Constants.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ public final class Constants extends BaseConstants {
public static final Integer MAX_INSERT_BATCH_SIZE = 1000;
public static final Integer INSERT_CONCURRENCY_LIMIT = 10;
public static final Integer MAX_INSERT_CONCURRENCY_LIMIT = 10;

public static final Integer DETOKENIZE_BATCH_SIZE = 50;
public static final Integer DETOKENIZE_CONCURRENCY_LIMIT = 10;
public static final Integer MAX_DETOKENIZE_BATCH_SIZE = 1000;
Expand Down
85 changes: 44 additions & 41 deletions v3/src/main/java/com/skyflow/utils/validations/Validations.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
package com.skyflow.utils.validations;

import java.util.ArrayList;
import java.util.List;

import com.skyflow.config.Credentials;
import com.skyflow.config.VaultConfig;
import com.skyflow.enums.InterfaceName;
Expand All @@ -14,47 +17,41 @@
import com.skyflow.vault.data.InsertRequest;
import com.skyflow.vault.data.TokenGroupRedactions;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

public class Validations extends BaseValidations {
private Validations() {
super();
}

public static void validateInsertRequest(InsertRequest insertRequest) throws SkyflowException {
String table = insertRequest.getTable();
ArrayList<InsertRecord> values = insertRequest.getRecords();
if (values == null) {
ArrayList<InsertRecord> records = insertRequest.getRecords();
if (records == null) {
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.VALUES_IS_REQUIRED.getLog(), InterfaceName.INSERT.getName()
ErrorLogs.RECORDS_IS_REQUIRED.getLog(), InterfaceName.INSERT.getName()
));
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.ValuesKeyError.getMessage());
} else if (values.isEmpty()) {
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.RecordsKeyError.getMessage());
} else if (records.isEmpty()) {
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.EMPTY_VALUES.getLog(), InterfaceName.INSERT.getName()
ErrorLogs.EMPTY_RECORDS.getLog(), InterfaceName.INSERT.getName()
));
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyValues.getMessage());
} else if (values.size() > 10000) {
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyRecords.getMessage());
} else if (records.size() > 10000) {
LogUtil.printErrorLog(ErrorLogs.RECORD_SIZE_EXCEED.getLog());
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.RecordSizeExceedError.getMessage());
}
// if (table == null) {
// LogUtil.printErrorLog(Utils.parameterizedString(
// ErrorLogs.TABLE_IS_REQUIRED.getLog(), InterfaceName.INSERT.getName()
// ));
// throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.TableKeyError.getMessage());
// } else if (table.trim().isEmpty()) {
// LogUtil.printErrorLog(Utils.parameterizedString(
// ErrorLogs.EMPTY_TABLE_NAME.getLog(), InterfaceName.INSERT.getName()
// ));
// throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyTable.getMessage());
// }
for (InsertRecord record : records) {
if(record == null){
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.INVALID_RECORD.getLog(), InterfaceName.INSERT.getName()
));
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.InvalidRecord.getMessage());
}
}

// table check if specified for both
if (insertRequest.getTable() != null && !table.trim().isEmpty()){ // if table name specified at both place
for (InsertRecord valuesMap : values) {
if (valuesMap.getTable() != null || !valuesMap.getTable().trim().isEmpty()){
for (InsertRecord record : records) {
if (record.getTable() != null && !record.getTable().trim().isEmpty()){
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.TABLE_SPECIFIED_AT_BOTH_PLACE.getLog(), InterfaceName.INSERT.getName()
));
Expand All @@ -64,8 +61,8 @@ public static void validateInsertRequest(InsertRequest insertRequest) throws Sky
}
// table check if not specified for both or if missing in any object
if (insertRequest.getTable() == null || table.trim().isEmpty()){ // if table name specified at both place
for (InsertRecord valuesMap : values) {
if (valuesMap.getTable() == null || valuesMap.getTable().trim().isEmpty()){
for (InsertRecord record : records) {
if (record.getTable() == null || record.getTable().trim().isEmpty()){
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.TABLE_NOT_SPECIFIED_AT_BOTH_PLACE.getLog(), InterfaceName.INSERT.getName()
));
Expand All @@ -75,8 +72,14 @@ public static void validateInsertRequest(InsertRequest insertRequest) throws Sky
}
// upsert check 1
if (insertRequest.getTable() != null && !table.trim().isEmpty()){ // if table name specified at both place
for (InsertRecord valuesMap : values) {
if (valuesMap.getUpsert() != null && !valuesMap.getUpsert().isEmpty()){
for (InsertRecord record : records) {
if (record.getUpsert() != null && record.getUpsert().isEmpty()) {
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.EMPTY_UPSERT_VALUES.getLog(), InterfaceName.INSERT.getName()
));
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyUpsertValues.getMessage());
}
if (record.getUpsert() != null && !record.getUpsert().isEmpty()){
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.UPSERT_TABLE_REQUEST_AT_RECORD_LEVEL.getLog(), InterfaceName.INSERT.getName()
));
Expand All @@ -94,24 +97,24 @@ public static void validateInsertRequest(InsertRequest insertRequest) throws Sky
}
}

// else if (upsert != null && upsert.isEmpty()) {
// LogUtil.printErrorLog(Utils.parameterizedString(
// ErrorLogs.EMPTY_UPSERT_VALUES.getLog(), InterfaceName.INSERT.getName()
// ));
// throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyUpsertValues.getMessage());
// }
if (insertRequest.getUpsert() != null && insertRequest.getUpsert().isEmpty()) {
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.EMPTY_UPSERT_VALUES.getLog(), InterfaceName.INSERT.getName()
));
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyUpsertValues.getMessage());
}

for (InsertRecord valuesMap : values) {
if (valuesMap != null ) {
if (valuesMap.getData() != null){
for (String key : valuesMap.getData().keySet()) {
for (InsertRecord record : records) {
if (record != null ) {
if (record.getData() != null){
for (String key : record.getData().keySet()) {
if (key == null || key.trim().isEmpty()) {
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.EMPTY_OR_NULL_KEY_IN_VALUES.getLog(), InterfaceName.INSERT.getName()
));
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyKeyInValues.getMessage());
throw new SkyflowException(ErrorCode.INVALID_INPUT.getCode(), ErrorMessage.EmptyKeyInRecords.getMessage());
} else {
Object value = valuesMap.getData().get(key);
Object value = record.getData().get(key);
if (value == null || value.toString().trim().isEmpty()) {
LogUtil.printErrorLog(Utils.parameterizedString(
ErrorLogs.EMPTY_OR_NULL_VALUE_IN_VALUES.getLog(),
Expand Down
5 changes: 3 additions & 2 deletions v3/src/main/java/com/skyflow/vault/data/DetokenizeResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,11 +9,12 @@
import java.util.stream.Collectors;

public class DetokenizeResponse {
@Expose(serialize = true)
private List<DetokenizeResponseObject> success;
@Expose(serialize = true)
private DetokenizeSummary summary;

@Expose(serialize = true)
private List<DetokenizeResponseObject> success;

@Expose(serialize = true)
private List<ErrorRecord> errors;

Expand Down
8 changes: 5 additions & 3 deletions v3/src/main/java/com/skyflow/vault/data/InsertRecord.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
package com.skyflow.vault.data;

import com.skyflow.enums.UpdateType;

import java.util.List;
import java.util.Map;

Expand All @@ -23,7 +25,7 @@ public List<String> getUpsert() {
return this.builder.upsert;
}

public String getUpsertType() {
public UpdateType getUpsertType() {
return this.builder.upsertType;
}

Expand All @@ -32,7 +34,7 @@ public static final class InsertRecordBuilder {
private String table;
private Map<String, Object> data;
private List<String> upsert;
private String upsertType;
private UpdateType upsertType;

public InsertRecordBuilder table(String table) {
this.table = table;
Expand All @@ -49,7 +51,7 @@ public InsertRecordBuilder upsert(List<String> upsert) {
return this;
}

public InsertRecordBuilder upsertType(String upsertType) {
public InsertRecordBuilder upsertType(UpdateType upsertType) {
this.upsertType = upsertType;
return this;
}
Expand Down
1 change: 1 addition & 0 deletions v3/src/main/java/com/skyflow/vault/data/Success.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ public Success(int index, String skyflow_id, Map<String, List<Token>> tokens, Ma
this.skyflow_id = skyflow_id;
this.tokens = tokens;
this.data = data;
this.table = table;
}

public String getSkyflowId() {
Expand Down
Loading
Loading