Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
123 changes: 122 additions & 1 deletion src/main/java/ai/chalk/client/ChalkClientImpl.java
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

import ai.chalk.exceptions.ChalkException;
import ai.chalk.exceptions.ClientException;
import ai.chalk.exceptions.ServerError;
import ai.chalk.internal.arrow.FeatherProcessor;
import ai.chalk.internal.bytes.BytesProducer;
import ai.chalk.internal.config.Loader;
Expand All @@ -13,11 +14,20 @@
import ai.chalk.internal.request.models.SendRequestParams;
import ai.chalk.models.OnlineQueryParamsComplete;
import ai.chalk.models.OnlineQueryResult;
import ai.chalk.models.QueryMeta;
import ai.chalk.models.UploadFeaturesParams;
import ai.chalk.models.UploadFeaturesResult;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategies;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.apache.arrow.memory.BufferAllocator;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.*;
import org.apache.arrow.vector.table.Table;

import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import java.util.*;

public class ChalkClientImpl implements ChalkClient {
Expand Down Expand Up @@ -58,6 +68,13 @@ public ChalkClientImpl(BuilderImpl config) throws ChalkException {
}

public OnlineQueryResult onlineQuery(OnlineQueryParamsComplete params) throws ChalkException {
if (BytesProducer.hasHasManyInputs(params)) {
return onlineQueryJson(params);
}
return onlineQueryFeather(params);
}

private OnlineQueryResult onlineQueryFeather(OnlineQueryParamsComplete params) throws ChalkException {
byte[] bodyBytes;
try (
var childAllocator = allocator.newChildAllocator(
Expand Down Expand Up @@ -88,11 +105,115 @@ public OnlineQueryResult onlineQuery(OnlineQueryParamsComplete params) throws Ch
0,
FeatherProcessor.ALLOCATOR_SIZE_RESPONSE
);
// ignore the warning here, because we don't want to free the memory yet
var bulkResponse = OnlineQueryBulkResponse.fromBytes(response.body(), allocator);
return bulkResponse.toResult();
}

private OnlineQueryResult onlineQueryJson(OnlineQueryParamsComplete params) throws ChalkException {
Map<String, Object> jsonBody = BytesProducer.buildJsonOnlineQueryBody(params);

SendRequestParams request = new SendRequestParams.Builder(params.getTimeout())
.path("/v1/query/online")
.body(jsonBody)
.method("POST")
.branch(params.getBranch())
.isEngineRequest(true)
.previewDeploymentId(params.getPreviewDeploymentId())
.environmentOverride(params.getEnvironmentId())
.queryName(params.getQueryName())
.build();

HttpResponse<byte[]> response = this.handler.sendRequest(request);

ObjectMapper mapper = new ObjectMapper();
mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
mapper.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
mapper.registerModule(new JavaTimeModule());

Map<String, Object> responseMap;
try {
responseMap = mapper.readValue(response.body(), Map.class);
} catch (Exception e) {
throw new ClientException("Failed to parse JSON query response", e);
}

ServerError[] errors = null;
if (responseMap.get("errors") instanceof List<?> errorsList && !errorsList.isEmpty()) {
errors = new ServerError[errorsList.size()];
for (int i = 0; i < errorsList.size(); i++) {
errors[i] = mapper.convertValue(errorsList.get(i), ServerError.class);
}
}

QueryMeta meta = null;
if (responseMap.get("meta") != null) {
meta = mapper.convertValue(responseMap.get("meta"), QueryMeta.class);
}

BufferAllocator childAllocator = this.allocator.newChildAllocator(
"online_query_json_response", 0, FeatherProcessor.ALLOCATOR_SIZE_RESPONSE
);

Table scalarTable = null;
if (responseMap.get("data") instanceof List<?> dataList && !dataList.isEmpty()) {
List<FieldVector> vectors = new ArrayList<>();
try {
for (Object item : dataList) {
var dataItem = (Map<String, Object>) item;
String fieldName = (String) dataItem.get("field");
Object value = dataItem.get("value");
if (fieldName == null) continue;

FieldVector vector = createVectorForValue(fieldName, value, childAllocator);
if (vector != null) vectors.add(vector);
}
scalarTable = new Table(vectors);
} catch (Exception e) {
for (FieldVector v : vectors) v.close();
childAllocator.close();
throw new ClientException("Failed to build Arrow table from JSON response", e);
}
}

return new OnlineQueryResult(scalarTable, new HashMap<>(), errors, meta, childAllocator);
}

private static FieldVector createVectorForValue(String name, Object value, BufferAllocator allocator) {
if (value == null) {
return new NullVector(name, 1);
} else if (value instanceof Integer intVal) {
BigIntVector v = new BigIntVector(name, allocator);
v.allocateNew(1);
v.set(0, intVal.longValue());
v.setValueCount(1);
return v;
} else if (value instanceof Long longVal) {
BigIntVector v = new BigIntVector(name, allocator);
v.allocateNew(1);
v.set(0, longVal);
v.setValueCount(1);
return v;
} else if (value instanceof Double doubleVal) {
Float8Vector v = new Float8Vector(name, allocator);
v.allocateNew(1);
v.set(0, doubleVal);
v.setValueCount(1);
return v;
} else if (value instanceof Boolean boolVal) {
BitVector v = new BitVector(name, allocator);
v.allocateNew(1);
v.set(0, boolVal ? 1 : 0);
v.setValueCount(1);
return v;
} else {
VarCharVector v = new VarCharVector(name, allocator);
v.allocateNew(1);
v.set(0, value.toString().getBytes(StandardCharsets.UTF_8));
v.setValueCount(1);
return v;
}
}

public UploadFeaturesResult uploadFeatures(UploadFeaturesParams params) throws ChalkException {
byte[] tableBytes;
try {
Expand Down
86 changes: 86 additions & 0 deletions src/main/java/ai/chalk/internal/bytes/BytesProducer.java
Original file line number Diff line number Diff line change
Expand Up @@ -84,9 +84,95 @@ public static Map<String, Object> buildJsonHeader(OnlineQueryParams params) {
}
jsonHeader.put("context", onlineQueryContext);

Map<String, String> hints = buildInputSchemaHint(params);
if (!hints.isEmpty()) {
jsonHeader.put("input_schema_hint", hints);
}

return jsonHeader;
}

/**
* Builds the input_schema_hint map by merging auto-detected has-many inputs
* with any user-provided hints. User-provided hints take precedence.
*/
static Map<String, String> buildInputSchemaHint(OnlineQueryParams params) {
Map<String, String> hints = new HashMap<>();

if (params.getInputs() != null) {
for (var entry : params.getInputs().entrySet()) {
String fqn = entry.getKey();
List<?> values = entry.getValue();
String hint = detectHasManyHint(fqn, values);
if (hint != null) {
hints.put(fqn, hint);
}
}
}

if (params.getInputSchemaHint() != null) {
hints.putAll(params.getInputSchemaHint());
}

return hints;
}

/**
* Detects whether an input value represents a has-many (list of structs/maps)
* and generates the bracketed projection hint string.
* Returns null if the input is not a has-many.
*/
private static String detectHasManyHint(String fqn, List<?> values) {
if (values == null || values.isEmpty()) {
return null;
}

// Scan all rows to find a non-empty inner list with Map elements
for (Object v : values) {
if (!(v instanceof List<?> innerList) || innerList.isEmpty()) {
continue;
}
for (Object item : innerList) {
if (item instanceof Map<?, ?> mapItem) {
var sb = new StringBuilder(fqn).append('[');
boolean first = true;
for (Object key : mapItem.keySet()) {
if (!first) sb.append(',');
sb.append(key);
first = false;
}
sb.append(']');
return sb.toString();
}
}
}
return null;
}

public static Map<String, Object> buildJsonOnlineQueryBody(OnlineQueryParamsComplete params) {
Map<String, Object> body = buildJsonHeader(params);

Map<String, Object> singleRowInputs = new HashMap<>();
for (var entry : params.getInputs().entrySet()) {
List<?> values = entry.getValue();
singleRowInputs.put(entry.getKey(), values.get(0));
}
body.put("inputs", singleRowInputs);
body.put("encoding_options", Map.of("encode_structs_as_objects", true));

return body;
}

public static boolean hasHasManyInputs(OnlineQueryParams params) {
if (params.getInputs() == null) return false;
for (List<?> values : params.getInputs().values()) {
if (values == null || values.isEmpty()) continue;
Object first = values.get(0);
if (first instanceof List<?>) return true;
}
return false;
}

public static byte[] convertOnlineQueryParamsToBytes(OnlineQueryParamsComplete params, BufferAllocator allocator) throws Exception {
byte[] arrowBytes;
if (params.getInputs() == null) {
Expand Down
2 changes: 2 additions & 0 deletions src/main/java/ai/chalk/internal/request/RequestHandler.java
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,8 @@ private byte[] getBodyBytes(Object body) throws Exception {
bodyBytes = (byte[]) body;
} else {
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.registerModule(new JavaTimeModule());
objectMapper.configure(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
objectMapper.setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
bodyBytes = objectMapper.writeValueAsBytes(body);
Expand Down
Loading
Loading