Add Flink sink connector + expose upload_features write targets

src/main/java/ai/chalk/client/GRPCClient.java

@@ -358,13 +358,29 @@ public UploadFeaturesResult uploadFeatures(UploadFeaturesParams params) throws C
             throw new ClientException("Failed to convert inputs to Arrow bytes", e);
         }
 
+        // Only set fields that deviate from the server defaults, mirroring the Python client
+        // (upload_features in client_grpc.py). write_online defaults to true server-side, so it is
+        // sent only when the caller opts out; this keeps prior callers wire-identical.
+        UploadFeaturesOptions.Builder optionsBuilder = UploadFeaturesOptions.newBuilder();
+        if (params.isUpdateMataggs()) {
+            optionsBuilder.setUpdateMataggs(true);
+        }
+        if (params.isWriteOffline()) {
+            optionsBuilder.setWriteOffline(true);
+        }
+        if (!params.isWriteOnline()) {
+            optionsBuilder.setWriteOnline(false);
+        }
+        UploadFeaturesOptions options = optionsBuilder.build();
+
         UploadFeaturesResponse response = this.stubsProvider.getQueryStub(Optional.ofNullable(params.getTimeout()))
             .withInterceptors(
                 this.getRequestHeaderInterceptor(params.getEnvironmentId(), null)
             )
             .uploadFeatures(
                 UploadFeaturesRequest.newBuilder()
                     .setInputsTable(ByteString.copyFrom(tableBytes))
+                    .setOptions(options)
                     .build()
             );
 

src/main/java/ai/chalk/models/UploadFeaturesParams.java

@@ -19,16 +19,30 @@ public class UploadFeaturesParams {
     private final String environmentId;
     @Nullable
     private final Duration timeout;
+    // Write targets (mirror chalk.common.v1.UploadFeaturesOptions). Defaults preserve prior
+    // behavior: online store only.
+    private final boolean writeOnline;
+    private final boolean writeOffline;
+    private final boolean updateMataggs;
 
     @AllArgsConstructor
     @NoArgsConstructor
     public static class Builder {
         protected Map<String, List<?>> inputs;
         protected String environmentId;
         protected Duration timeout;
+        protected boolean writeOnline = true;
+        protected boolean writeOffline = false;
+        protected boolean updateMataggs = false;
 
         public UploadFeaturesParams build() {
-            return new UploadFeaturesParams(this.inputs, this.environmentId, this.timeout);
+            return new UploadFeaturesParams(
+                this.inputs,
+                this.environmentId,
+                this.timeout,
+                this.writeOnline,
+                this.writeOffline,
+                this.updateMataggs);
         }
 
         private void initInputs() {
@@ -67,6 +81,24 @@ public Builder withTimeout(Duration timeout) {
             this.timeout = timeout;
             return this;
         }
+
+        /** Write uploaded values to the online store. Defaults to {@code true}. */
+        public Builder withWriteOnline(boolean writeOnline) {
+            this.writeOnline = writeOnline;
+            return this;
+        }
+
+        /** Also write uploaded values to the offline store. Defaults to {@code false}. */
+        public Builder withWriteOffline(boolean writeOffline) {
+            this.writeOffline = writeOffline;
+            return this;
+        }
+
+        /** Update materialized (streaming) aggregations on upload. Defaults to {@code false}. */
+        public Builder withUpdateMataggs(boolean updateMataggs) {
+            this.updateMataggs = updateMataggs;
+            return this;
+        }
     }
 
     public static Builder builder() {