Fix ScriptImplementor projection when using filters and add many tests (#136)

* Add some missing error handling and tests around errors

* Add tests for venice primitive keys

* Pass through missing properties

* Fix issues with top-level filter passing through wrong fields to ScriptImplementor

Author: jogrogan

Makefile

@@ -83,8 +83,8 @@ undeploy-kafka:
 # Deploys Venice cluster in docker and creates two stores in Venice. Stores are not managed via K8s for now.
 deploy-venice: deploy deploy-flink
 	docker compose -f ./deploy/docker/venice/docker-compose-single-dc-setup.yaml up -d --wait
-	docker exec venice-client ./create-store.sh http://venice-controller:5555 venice-cluster0 test-store schemas/keySchema.avsc schemas/valueSchema.avsc
-	docker exec venice-client ./create-store.sh http://venice-controller:5555 venice-cluster0 test-store-1 schemas/keySchema.avsc schemas/valueSchema.avsc
+	docker exec venice-client ./create-store.sh http://venice-controller:5555 venice-cluster0 test-store schemas/keySchemaRecord.avsc schemas/valueSchema.avsc
+	docker exec venice-client ./create-store.sh http://venice-controller:5555 venice-cluster0 test-store-primitive schemas/keySchemaPrimitive.avsc schemas/valueSchema.avsc
 	kubectl apply -f ./deploy/samples/venicedb.yaml
 
 undeploy-venice:
@@ -101,7 +101,7 @@ integration-tests: deploy-dev-environment
 	kubectl port-forward -n kafka svc/one-kafka-external-bootstrap 9092 & echo $$! > port-forward.pid
 	kubectl port-forward -n flink svc/flink-sql-gateway 8083 & echo $$! > port-forward-2.pid
 	kubectl port-forward -n flink svc/basic-session-deployment-rest 8081 & echo $$! > port-forward-3.pid
-	./gradlew intTest --no-parallel || kill `cat port-forward.pid port-forward-2.pid, port-forward-3.pid`
+	./gradlew intTest --no-parallel || kill `cat port-forward.pid port-forward-2.pid port-forward-3.pid`
 	kill `cat port-forward.pid`
 	kill `cat port-forward-2.pid`
 	kill `cat port-forward-3.pid`

deploy/docker/venice/schemas/keySchemaPrimitive.avsc

@@ -0,0 +1 @@
+"int"

deploy/docker/venice/schemas/keySchema.avsc renamed to deploy/docker/venice/schemas/keySchemaRecord.avsc

@@ -65,6 +65,7 @@
 import com.linkedin.hoptimator.Pipeline;
 import com.linkedin.hoptimator.View;
 import com.linkedin.hoptimator.util.DeploymentService;
+import com.linkedin.hoptimator.util.planner.HoptimatorJdbcTable;
 import com.linkedin.hoptimator.util.planner.PipelineRel;
 
 
@@ -99,7 +100,14 @@ public DdlExecutor getDdlExecutor() {
   @Override
   public void execute(SqlCreateView create, CalcitePrepare.Context context) {
     final Pair<CalciteSchema, String> pair = schema(context, true, create.name);
+    if (pair.left == null) {
+      throw new DdlException(create, "Schema for " + create.name + " not found.");
+    }
     final SchemaPlus schemaPlus = pair.left.plus();
+    if (schemaPlus.getTable(pair.right) instanceof HoptimatorJdbcTable) {
+      throw new DdlException(create,
+          "Cannot overwrite physical table " + pair.right + " with a view.");
+    }
     for (Function function : schemaPlus.getFunctions(pair.right)) {
       if (function.getParameters().isEmpty()) {
         if (!create.getReplace()) {
@@ -139,17 +147,22 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
   public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context context) {
     final Pair<CalciteSchema, String> pair = schema(context, true, create.name);
     if (pair.left == null) {
-      throw new DdlException(create, "Schema for " + create.name.getSimple() + " not found.");
+      throw new DdlException(create, "Schema for " + create.name + " not found.");
     }
-    if (pair.left.plus().getTable(pair.right) != null) {
+    final SchemaPlus schemaPlus = pair.left.plus();
+    if (schemaPlus.getTable(pair.right) != null) {
+      if (schemaPlus.getTable(pair.right) instanceof HoptimatorJdbcTable) {
+        throw new DdlException(create,
+            "Cannot overwrite physical table " + pair.right + " with a view.");
+      }
       // Materialized view exists.
       if (!create.ifNotExists && !create.getReplace()) {
         // They did not specify IF NOT EXISTS, so give error.
         throw new DdlException(create,
             "View " + pair.right + " already exists. Use CREATE OR REPLACE to update.");
       }
       if (create.getReplace()) {
-        pair.left.plus().removeTable(pair.right);
+        schemaPlus.removeTable(pair.right);
       } else {
         // nothing to do
         return;
@@ -160,7 +173,6 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
     final String sql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
     final List<String> schemaPath = pair.left.path(null);
 
-    SchemaPlus schemaPlus = pair.left.plus();
     String schemaName = schemaPlus.getName();
     String viewName = pair.right;
     List<String> viewPath = new ArrayList<>(schemaPath);
@@ -190,7 +202,7 @@ public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context con
       connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, pipelineName);
       List<String> sinkPath = new ArrayList<>(schemaPath);
       sinkPath.add(sinkName);
-      Table sink = pair.left.plus().getTable(sinkName);
+      Table sink = schemaPlus.getTable(sinkName);
 
       final RelDataType rowType;
       if (sink != null) {

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlExecutor.java

@@ -80,7 +80,7 @@ public void execute(Context context, boolean execute) throws Exception {
                 String sql = context.previousSqlCommand().sql;
                 HoptimatorConnection conn = (HoptimatorConnection) connection;
                 RelRoot root = HoptimatorDriver.convert(conn, sql).root;
-                String []parts = line.split(" ", 2);
+                String[] parts = line.split(" ", 2);
                 String pipelineName = parts.length == 2 ? parts[1] : "test";
                 Pipeline pipeline = DeploymentService.plan(root, Collections.emptyList(), conn.connectionProperties())
                     .pipeline(pipelineName, conn.connectionProperties());

hoptimator-jdbc/src/test/resources/validation.id

@@ -19,4 +19,10 @@ public void k8sDdlScript() throws Exception {
   public void k8sDdlScriptFunction() throws Exception {
     run("k8s-ddl-function.id", "fun=mysql");
   }
+
+  @Test
+  @Tag("integration")
+  public void k8sValidationScript() throws Exception {
+    run("k8s-validation.id");
+  }
 }

hoptimator-jdbc/src/test/resources/views.id

@@ -88,6 +88,41 @@ select * from ads.target2;
 
 !ok
 
+create or replace materialized view ads."PAGE_VIEWS$filter" as select first_name as page_urn, last_name as member_urn from profile.members where first_name = 'Alice';
+(0 rows modified)
+
+!update
+
+select * from ads."PAGE_VIEWS$filter";
++----------+------------+
+| PAGE_URN | MEMBER_URN |
++----------+------------+
+| Alice    | Addison    |
++----------+------------+
+(1 row)
+
+!ok
+
+create or replace materialized view ads."PAGE_VIEWS$filter" (member_urn) as select last_name as member_urn from profile.members where last_name = 'Addison';
+(0 rows modified)
+
+!update
+
+select * from ads."PAGE_VIEWS$filter";
++------------+
+| MEMBER_URN |
++------------+
+| Addison    |
++------------+
+(1 row)
+
+!ok
+
+drop materialized view ads."PAGE_VIEWS$filter";
+(0 rows modified)
+
+!update
+
 drop materialized view ads.target2;
 (0 rows modified)
 
@@ -106,4 +141,25 @@ drop materialized view ads.audience2;
 drop view ads.audience;
 (0 rows modified)
 
-!update
+!update
+
+insert into ads.page_views select first_name as page_urn, last_name as member_urn from profile.members where first_name = 'Alice';
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkSessionJob
+metadata:
+  name: ads-database-pageviews
+spec:
+  deploymentName: basic-session-deployment
+  job:
+    entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
+    args:
+    - CREATE DATABASE IF NOT EXISTS `PROFILE` WITH ()
+    - CREATE TABLE IF NOT EXISTS `PROFILE`.`MEMBERS` (`FIRST_NAME` VARCHAR, `LAST_NAME` VARCHAR, `MEMBER_URN` VARCHAR, `COMPANY_URN` VARCHAR) WITH ('connector'='datagen', 'number-of-rows'='10')
+    - CREATE DATABASE IF NOT EXISTS `ADS` WITH ()
+    - CREATE TABLE IF NOT EXISTS `ADS`.`PAGE_VIEWS` (`PAGE_URN` VARCHAR, `MEMBER_URN` VARCHAR) WITH ('connector'='blackhole')
+    - INSERT INTO `ADS`.`PAGE_VIEWS` (`PAGE_URN`, `MEMBER_URN`) SELECT `FIRST_NAME` AS `PAGE_URN`, `LAST_NAME` AS `MEMBER_URN` FROM `PROFILE`.`MEMBERS` WHERE `FIRST_NAME` = 'Alice'
+    jarURI: file:///opt/hoptimator-flink-runner.jar
+    parallelism: 1
+    upgradeMode: stateless
+    state: running
+!specify PAGE_VIEWS

hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

@@ -0,0 +1,36 @@
+!set outputformat mysql
+!use k8s
+
+create or replace materialized view ads.page_views as select first_name, last_name from profile.members;
+Cannot overwrite physical table PAGE_VIEWS with a view.
+!error
+
+create or replace materialized view invalid.page_views as select first_name, last_name from profile.members;
+Schema for INVALID.PAGE_VIEWS not found.
+!error
+
+create or replace materialized view ads.audience as select first_name, last_name from ads.page_views natural join profile.members;
+(0 rows modified)
+
+!update
+
+create materialized view ads.audience as select first_name, last_name from ads.page_views natural join profile.members;
+View AUDIENCE already exists. Use CREATE OR REPLACE to update.
+!error
+
+drop materialized view ads.audience;
+(0 rows modified)
+
+!update
+
+create or replace materialized view ads."PAGE_VIEWS$myview" as select first_name from profile.members;
+Field FIRST_NAME not found in sink schema
+!error
+
+create or replace materialized view ads."PAGE_VIEWS$myview" (page_urn, member_urn) as select first_name from profile.members;
+List of column aliases must have same degree as table; table has 1 columns ('FIRST_NAME'), whereas alias list has 2 columns
+!error
+
+create or replace materialized view ads."PAGE_VIEWS$myview" (page_urn) as select first_name, last_name from profile.members;
+List of column aliases must have same degree as table; table has 2 columns ('FIRST_NAME', 'LAST_NAME'), whereas alias list has 1 columns
+!error