Add Venice support #78

jogrogan · 2024-12-17T17:21:47Z

Adds Venice support to Hoptimator.

Queries Venice for Key & Value Avro schemas, merges them into one schema, keys structured as KEY$ to prevent collisions
Passes KEY fields through to Sink options (intended to be used by flink under key.fields connector property)
Supports partial inserts of the form insert into "VENICE-CLUSTER0"."test-store-1" ("KEY$id", "stringField") SELECT ...
- Fields not specified in the insert are required to be nullable (this may still be a problem for fields that are not nullable but do have a default in Avro)
- Captures the targetFields specified in the insert, and drops other fields.
- Rewrites INSERT with aliasing for fields specified (see examples in comments around ScriptImplementor)

Other changes in this PR:

Adds Venice tests
Adds ability to stand up Venice locally in Docker
Clean up Makefile targets
Address various checkstyle errors

Implemented the Venice driver/schema classes with separate overridable functions to be able to handle company-internal connection components via a simple override

./hoptimator
0: Hoptimator> create or replace materialized view "VENICE-CLUSTER0"."my-store" as select * from "VENICE-CLUSTER0"."test-store";

0: Hoptimator> !tables
+-----------+-----------------+------------------+-------------------+---------+
| TABLE_CAT |   TABLE_SCHEM   |    TABLE_NAME    |    TABLE_TYPE     | REMARKS |
+-----------+-----------------+------------------+-------------------+---------+
...
|           | VENICE-CLUSTER0 | my-store         | MATERIALIZED VIEW |         |
|           | VENICE-CLUSTER0 | test-store       | TABLE             |         |
|           | VENICE-CLUSTER0 | test-store-1     | TABLE             |         |
...
+-----------+-----------------+------------------+-------------------+---------+

$ k get flinkdeployments.flink.apache.org -o yaml venice-cluster0-my-store
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  creationTimestamp: "2024-12-24T20:38:41Z"
  finalizers:
  - flinkdeployments.flink.apache.org/finalizer
  generation: 2
  name: venice-cluster0-my-store
  namespace: default
  resourceVersion: "32186"
  uid: e0d855d3-af15-49d9-a16b-13c87579a23e
spec:
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "1"
  flinkVersion: v1_16
  image: docker.io/library/hoptimator-flink-runner
  imagePullPolicy: Never
  job:
    args:
    - CREATE TABLE IF NOT EXISTS `test-store` (`intField` INTEGER, `stringField` VARCHAR,
      `KEY` ROW(`id` INTEGER)) WITH ('connector'='venice', 'key.fields'='KEY_id',
      'key.fields-prefix'='KEY_', 'partial-update-mode'='true', 'storeName'='test-store',
      'value.fields-include'='EXCEPT_KEY');
    - CREATE TABLE IF NOT EXISTS `my-store` (`intField` INTEGER, `stringField` VARCHAR,
      `KEY_id` INTEGER) WITH ('connector'='venice', 'key.fields'='KEY_id', 'key.fields-prefix'='KEY_',
      'partial-update-mode'='true', 'storeName'='my-store', 'value.fields-include'='EXCEPT_KEY');
    - INSERT INTO `my-store` (`intField`, `stringField`, `KEY_id`) SELECT * FROM `VENICE-CLUSTER0`.`test-store`;
    entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
    jarURI: local:///opt/hoptimator-flink-runner.jar
    parallelism: 1
    state: running
    upgradeMode: stateless
  jobManager:
    replicas: 1
    resource:
      cpu: 0.1
      memory: 2048m
  serviceAccount: flink
  taskManager:
    resource:
      cpu: 0.1
      memory: 2048m
status:
  clusterInfo: {}
  jobManagerDeploymentStatus: DEPLOYING
  jobStatus:
    checkpointInfo:
      lastPeriodicCheckpointTimestamp: 0
    jobId: 49de826b428ba2d93e4fac8b120de2c4
    savepointInfo:
      lastPeriodicSavepointTimestamp: 0
      savepointHistory: []
    state: RECONCILING
  lifecycleState: DEPLOYED
  observedGeneration: 2
  reconciliationStatus:
    lastReconciledSpec: '{"spec":{"job":{"jarURI":"local:///opt/hoptimator-flink-runner.jar","parallelism":1,"entryClass":"com.linkedin.hoptimator.flink.runner.FlinkRunner","args":["CREATE
      TABLE IF NOT EXISTS `test-store` (`intField` INTEGER, `stringField` VARCHAR,
      `KEY` ROW(`id` INTEGER)) WITH (''connector''=''venice'', ''key.fields''=''KEY_id'',
      ''key.fields-prefix''=''KEY_'', ''partial-update-mode''=''true'', ''storeName''=''test-store'',
      ''value.fields-include''=''EXCEPT_KEY'');","CREATE TABLE IF NOT EXISTS `my-store`
      (`intField` INTEGER, `stringField` VARCHAR, `KEY_id` INTEGER) WITH (''connector''=''venice'',
      ''key.fields''=''KEY_id'', ''key.fields-prefix''=''KEY_'', ''partial-update-mode''=''true'',
      ''storeName''=''my-store'', ''value.fields-include''=''EXCEPT_KEY'');","INSERT
      INTO `my-store` (`intField`, `stringField`, `KEY_id`) SELECT * FROM `VENICE-CLUSTER0`.`test-store`;"],"state":"running","savepointTriggerNonce":null,"initialSavepointPath":null,"checkpointTriggerNonce":null,"upgradeMode":"stateless","allowNonRestoredState":null,"savepointRedeployNonce":null},"restartNonce":null,"flinkConfiguration":{"taskmanager.numberOfTaskSlots":"1"},"image":"docker.io/library/hoptimator-flink-runner","imagePullPolicy":"Never","serviceAccount":"flink","flinkVersion":"v1_16","ingress":null,"podTemplate":null,"jobManager":{"resource":{"cpu":0.1,"memory":"2048m","ephemeralStorage":null},"replicas":1,"podTemplate":null},"taskManager":{"resource":{"cpu":0.1,"memory":"2048m","ephemeralStorage":null},"replicas":null,"podTemplate":null},"logConfiguration":null,"mode":null},"resource_metadata":{"apiVersion":"flink.apache.org/v1beta1","metadata":{"generation":2},"firstDeployment":true}}'
    reconciliationTimestamp: 1735072722874
    state: DEPLOYED
  taskManager:
    labelSelector: component=taskmanager,app=venice-cluster0-my-store
    replicas: 1

See included tests for more samples

hoptimator-kafka/src/test/resources/kafka-ddl.id

jogrogan · 2024-12-17T17:24:02Z

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/PipelineRel.java

+      this.sinkOptions = addKeysAsOption(options, rowType);
+    }
+
+    private Map<String, String> addKeysAsOption(Map<String, String> options, RelDataType rowType) {


I don't love this approach, open to suggestions.

I looked into hints to solve this and did get them working to an extent (will open a separate PR) but this would require users to pass in key information into their SQL statement. I have not figured out a way to inject hints at runtime from VeniceDriver.

I guess I'm surprised we need to fully specify the keys in the options. The Kafka connector has similar properties (key.prefix, key.fields), but you don't need both. Is the Venice connector doing something different here? I'd expect key.prefix=key_ to be sufficient.

Also, how would the Venice connector behave if we grouped the keys in a Row(...) object? Can we just have key.fields=KEY and then KEY ROW(F1 VARCHAR, F2 INT) etc?

(Not suggesting we do that, just asking if possible?)

I guess I'm surprised we need to fully specify the keys in the options. The Kafka connector has similar properties (key.prefix, key.fields), but you don't need both. Is the Venice connector doing something different here? I'd expect key.prefix=key_ to be sufficient.

Yea I confirmed it is an issue with Venice due do some additional avro schema validation they do. They pull the keySchema and validate it against key.fields (separate from the prefix). The key.prefix allow these names to be different like "id" vs "key_id"

Looking into the ROW syntax and it doesn't seem that is possible in Flink, there is no way to get Flink to destruct that ROW

What does retaining partitioning mean for the Kafka -> Venice use case? It seems like it is more of a problem on the producer side. Users that expect the same partitioning behavior would have to key their Kafka topic using the same combination of keys as Venice. We did this in Brooklin by constructing the producer key as a simple string with key values separated by _ from the source keys. Of course this isn't the same as identity partitioning but it does ensure that downstream consumer tasks read the same combo of keys.

Even for the Kafka -> Kafka use case, we aren't the ones consuming, the partitioning behavior comes from Flink right? I haven't looked into it to be fair, not sure how the behavior changes if you define key.fields there.

In the Kafka -> Kafka use case, insert into kafka.foo select * from kafka.bar intuitively keeps the key the same, since nothing in the SQL explicitly specifies otherwise. The effect should be that foo is a mirror of bar, with the same key and partition semantics. This is indeed the case today (internally). To the SQL engine, the input table always has an implicit KEY column, which gets selected by select *.

A typical use case for Kafka->Kafka pipelines is actually dropping the key and re-partitioning as round-robin. This is done today via SELECT *, NULL AS KEY.... Hoptimator explicitly supports NULL AS KEY, since it is not really possible to express "select everything except the key" in SQL.

I'm less familiar with the Venice use case, but I want to make sure that both Kafka->Kafka and Kafka->Venice are intuitive and somewhat consistent.

Think I get what you are saying. The Venice connector as written today doesn't look like it supports the same implicit keys. It relies on explicit keys via the key.fields table option.

Kafka->Kafka does support this today, you could have something like:
Kafka topic bar has a payload containing two fields id & val.
Specifying a flink table option of "key.fields": "id" before applying insert into kafka.foo select * from kafka.bar will use the column id as the key for topic foo when doing Kafka->Kafka. If there is an actual key present in the topic bar I don't believe it'll actually get used anywhere, it'll effectively be dropped.

A simple statement like insert into venice.foo select * from kafka.bar will still work as long as kafka.bar contains all the columns Venice expects, both key & value columns.

Currently select * from kafka.bar will always include a KEY VARCHAR, since internally all our Kafka topics are keyed with a simple string (or null). This new Venice integration expects a KEY field, which contains all the keys. Individually, both make sense. I'm just wondering if KEY VARCHAR and KEY Row(A ..., B ...) need some additional magic to work interchangeably. In particular, would Kafka->Venice end up with key.fields = KEY_KEY? Or can we detect that there is only one key and have key.fields = KEY?

Actually, I think what you have here with KEY_ as a magic prefix is going to be more robust than what we currently do for Kafka. Rather than try to shoehorn our existing simplistic Kafka convention here, let's adopt your approach for Kafka->Kafka as well. I think the only thing we'd need to do is change Kafka's magic column from KEY to KEY_STRING or something, and your code will just work. Kafka keys will just pop out as KEY Row(STRING VARCHAR) and the connector will see key.fields = KEY_STRING.

hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceStore.java

ryannedolan · 2024-12-17T22:35:59Z

Makefile

@@ -9,8 +9,8 @@ build:

 bounce: build undeploy deploy deploy-samples deploy-config deploy-demo

-# Integration tests expect K8s and Kafka to be running
-integration-tests: deploy-dev-environment deploy-samples
+# Integration tests expect K8s, Kafka, and Venice to be running


🔥 🔥 🔥

hoptimator-kafka/src/test/resources/kafka-ddl.id

ryannedolan · 2024-12-17T22:41:11Z

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/PipelineRel.java

+      this.sinkOptions = addKeysAsOption(options, rowType);
+    }
+
+    private Map<String, String> addKeysAsOption(Map<String, String> options, RelDataType rowType) {


I guess I'm surprised we need to fully specify the keys in the options. The Kafka connector has similar properties (key.prefix, key.fields), but you don't need both. Is the Venice connector doing something different here? I'd expect key.prefix=key_ to be sufficient.

ryannedolan · 2024-12-17T22:46:17Z

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/PipelineRel.java

+      this.sinkOptions = addKeysAsOption(options, rowType);
+    }
+
+    private Map<String, String> addKeysAsOption(Map<String, String> options, RelDataType rowType) {


Also, how would the Venice connector behave if we grouped the keys in a Row(...) object? Can we just have key.fields=KEY and then KEY ROW(F1 VARCHAR, F2 INT) etc?

(Not suggesting we do that, just asking if possible?)

ryannedolan · 2024-12-23T20:25:39Z

deploy/docker/venice/keySchema.avsc

@@ -0,0 +1,11 @@
+{
+  "type": "record",


We need to get create table venice.foo working :)

Agreed, I will spend some time looking into this when I can. Should be a simple API call just as I'm doing to fetch schemas, just different than the current paradigm since it isn't managed via K8s.

hoptimator-avro/build.gradle

ryannedolan · 2024-12-23T20:31:19Z

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/ScriptImplementor.java

+  // Without forced projection this will get optimized to:
+  // INSERT INTO `my-store` (`KEYFIELD`, `VARCHARFIELD`) SELECT * FROM `KAFKA`.`existing-topic-1`;
+  // With forced project this will resolve as:
+  // INSERT INTO `my-store` (`KEY_id`, `stringField`) SELECT `KEYFIELD` AS `KEY_id`, \


hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/ScriptImplementor.java

hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/LocalControllerClient.java

hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceDriver.java

hoptimator-venice/src/main/java/com/linkedin/hoptimator/venice/VeniceStore.java

ryannedolan

The if (schema.startsWith("VENICE")...) logic needs to be fixed, but I think we can accept the TODO and fix later.

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/PipelineRel.java

jogrogan commented Dec 17, 2024

View reviewed changes

hoptimator-kafka/src/test/resources/kafka-ddl.id Outdated Show resolved Hide resolved

jogrogan commented Dec 17, 2024

View reviewed changes

jogrogan force-pushed the jogrogan/venice branch from 6128407 to 4c2ffb9 Compare December 17, 2024 17:42

ryannedolan reviewed Dec 17, 2024

View reviewed changes