Fix field projection when all fields are specified (#190)

* Fix issue when all fields are specified but not SELECT *

* Make Quidem the same as hoptimator cli & fix tests

Author: jogrogan

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

@@ -1,5 +1,7 @@
 package com.linkedin.hoptimator.jdbc;
 
+import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
+import com.linkedin.hoptimator.util.planner.PipelineRel;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
@@ -20,7 +22,11 @@
 import java.util.Properties;
 import java.util.stream.Collectors;
 
+import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.dialect.CalciteSqlDialect;
 import org.junit.jupiter.api.Assertions;
 
 import net.hydromatic.quidem.AbstractCommand;
@@ -88,14 +94,38 @@ 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 pipelineName = parts.length == 2 ? parts[1] : "test";
-              Properties properties = new Properties();
-              properties.putAll(conn.connectionProperties());
-              properties.put(DeploymentService.PIPELINE_OPTION, pipelineName);
-              Pipeline pipeline = DeploymentService.plan(root, Collections.emptyList(), properties)
-                  .pipeline(pipelineName, conn);
+              String viewName = parts.length == 2 ? parts[1] : "test";
+              String querySql = sql;
+              SqlCreateMaterializedView create = null;
+              SqlNode sqlNode = HoptimatorDriver.parseQuery(conn, sql);
+              if (sqlNode.getKind().belongsTo(SqlKind.DDL)) {
+                if (sqlNode instanceof SqlCreateMaterializedView) {
+                  create = (SqlCreateMaterializedView) sqlNode;
+                  final SqlNode q = HoptimatorDdlUtils.renameColumns(create.columnList, create.query);
+                  querySql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
+                  viewName = HoptimatorDdlUtils.viewName(create.name);
+                } else {
+                  throw new RuntimeException("Unsupported DDL statement.");
+                }
+              }
+
+              RelRoot root = HoptimatorDriver.convert(conn, querySql).root;
+              Properties connectionProperties = conn.connectionProperties();
+              RelOptTable table = root.rel.getTable();
+              if (table != null) {
+                connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, String.join(".", table.getQualifiedName()));
+              } else if (create != null) {
+                connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, create.name.toString());
+              }
+
+              PipelineRel.Implementor plan = DeploymentService.plan(root, Collections.emptyList(), connectionProperties);
+              if (create != null) {
+                HoptimatorDdlUtils.snapshotAndSetSinkSchema(conn.createPrepareContext(),
+                    new HoptimatorDriver.Prepare(conn), plan, create, querySql);
+              }
+              Pipeline pipeline = plan.pipeline(viewName, conn);
               List<String> specs = new ArrayList<>();
               for (Source source : pipeline.sources()) {
                 specs.addAll(DeploymentService.specify(source, conn));

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

@@ -203,3 +203,45 @@ spec:
     upgradeMode: stateless
     state: running
 !specify AD_CLICKS
+
+create or replace materialized view ads."PAGE_VIEWS$forward-field-order" as select CAMPAIGN_URN as PAGE_URN, MEMBER_URN as MEMBER_URN from ads.ad_clicks;
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkSessionJob
+metadata:
+  name: ads-database-pageviews-forward-field-order
+spec:
+  deploymentName: basic-session-deployment
+  job:
+    entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
+    args:
+    - CREATE DATABASE IF NOT EXISTS `ADS` WITH ()
+    - CREATE TABLE IF NOT EXISTS `ADS`.`AD_CLICKS` (`CAMPAIGN_URN` VARCHAR, `MEMBER_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 * FROM `ADS`.`AD_CLICKS`
+    jarURI: file:///opt/hoptimator-flink-runner.jar
+    parallelism: 1
+    upgradeMode: stateless
+    state: running
+!specify PAGE_VIEWS
+
+create or replace materialized view ads."PAGE_VIEWS$reverse-field-order" as select MEMBER_URN as MEMBER_URN, CAMPAIGN_URN as PAGE_URN from ads.ad_clicks;
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkSessionJob
+metadata:
+  name: ads-database-pageviews-reverse-field-order
+spec:
+  deploymentName: basic-session-deployment
+  job:
+    entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
+    args:
+    - CREATE DATABASE IF NOT EXISTS `ADS` WITH ()
+    - CREATE TABLE IF NOT EXISTS `ADS`.`AD_CLICKS` (`CAMPAIGN_URN` VARCHAR, `MEMBER_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` (`MEMBER_URN`, `PAGE_URN`) SELECT `MEMBER_URN`, `CAMPAIGN_URN` AS `PAGE_URN` FROM `ADS`.`AD_CLICKS`
+    jarURI: file:///opt/hoptimator-flink-runner.jar
+    parallelism: 1
+    upgradeMode: stateless
+    state: running
+!specify PAGE_VIEWS

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

@@ -12,13 +12,13 @@ spec:
   sql:
     - PLACEHOLDER
   configs:
-    source.group.id: hoptimator-flink-beam-existing-topic-1
+    source.group.id: hoptimator-flink-beam-KAFKA.existing-topic-1
     source.schemas: KAFKA
     source.tables: existing-topic-2
     sink.schema: KAFKA
     sink.table: existing-topic-1
     fields: '{"VALUE":"VALUE","KEY":"KEY"}'
-    pipeline: 'existing-topic-1'
+    pipeline: 'KAFKA.existing-topic-1'
     flink.app.type: 'BEAM'
     flink.app.name: 'hoptimator-flink-runner'
 ---

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

@@ -12,6 +12,7 @@
 
 import javax.annotation.Nullable;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Project;
 import org.apache.calcite.rel.core.RelFactories;
 import org.apache.calcite.rel.rel2sql.RelToSqlConverter;
 import org.apache.calcite.rel.rel2sql.SqlImplementor;
@@ -347,7 +348,16 @@ public void implement(SqlWriter w) {
       String effectiveTable = suffix != null ? table + suffix : table;
       (new CompoundIdentifierImplementor(catalog, schema, effectiveTable)).implement(w);
       RelNode project = dropFields(relNode, targetFields);
-      (new ColumnListImplementor(project.getRowType())).implement(w);
+
+      // If the relNode is a Project (or subclass), the field names should already match the sink.
+      // Otherwise, like in SELECT * situations, the relNode fields will match the source field names, so
+      // we need to directly use targetFields to map correctly.
+      if (relNode instanceof Project) {
+        (new ColumnListImplementor(project.getRowType().getFieldNames())).implement(w);
+      } else {
+        (new ColumnListImplementor(targetFields.rightList())).implement(w);
+      }
+
       (new QueryImplementor(project, tableNameReplacements)).implement(w);
       w.literal(";");
     }
@@ -356,15 +366,35 @@ public void implement(SqlWriter w) {
     // Drops non-target columns, for use case: INSERT INTO (col1, col2) SELECT * FROM ...
     private static RelNode dropFields(RelNode relNode, ImmutablePairList<Integer, String> targetFields) {
       List<Integer> cols = new ArrayList<>();
-      int i = 0;
       List<String> targetFieldNames = targetFields.rightList();
-      for (RelDataTypeField field : relNode.getRowType().getFieldList()) {
-        if (targetFieldNames.contains(field.getName())
-            && !field.getType().getSqlTypeName().equals(SqlTypeName.NULL)) {
-          cols.add(i);
+      List<RelDataTypeField> sourceFields = relNode.getRowType().getFieldList();
+
+      // If the relNode is a Project (or subclass), the field names should already match the target
+      // because the projection explicitly renamed them. Use name-based matching.
+      if (relNode instanceof Project) {
+        for (int i = 0; i < sourceFields.size(); i++) {
+          RelDataTypeField field = sourceFields.get(i);
+          if (targetFieldNames.contains(field.getName())
+              && !field.getType().getSqlTypeName().equals(SqlTypeName.NULL)) {
+            cols.add(i);
+          }
+        }
+
+        return createForceProject(relNode, cols);
+      }
+
+      // Otherwise (e.g., TableScan), the projection was optimized away.
+      // Use index-based matching from targetFields to select the right columns.
+      for (int i = 0; i < targetFields.size(); i++) {
+        int fieldIndex = targetFields.leftList().get(i);
+        if (fieldIndex < sourceFields.size()) {
+          RelDataTypeField field = sourceFields.get(fieldIndex);
+          if (!field.getType().getSqlTypeName().equals(SqlTypeName.NULL)) {
+            cols.add(fieldIndex);
+          }
         }
-        i++;
       }
+
       return createForceProject(relNode, cols);
     }
   }
@@ -605,25 +635,20 @@ private static SqlDataTypeSpec maybeNullable(RelDataType dataType, SqlDataTypeSp
 
   /** Implements column lists, e.g. `NAME, AGE` */
   class ColumnListImplementor implements ScriptImplementor {
-    private final List<RelDataTypeField> fields;
+    private final List<String> fieldNames;
 
-    public ColumnListImplementor(RelDataType dataType) {
-      this(dataType.getFieldList());
-    }
-
-    public ColumnListImplementor(List<RelDataTypeField> fields) {
-      this.fields = fields;
+    ColumnListImplementor(List<String> fieldNames) {
+      this.fieldNames = fieldNames;
     }
 
     @Override
     public void implement(SqlWriter w) {
       SqlWriter.Frame frame1 = w.startList("(", ")");
-      List<SqlIdentifier> fieldNames = fields.stream()
-          .map(RelDataTypeField::getName)
+      List<SqlIdentifier> identifiers = fieldNames.stream()
           .map(x -> x.replaceAll("\\$", "_"))  // support FOO$BAR columns as FOO_BAR
           .map(x -> new SqlIdentifier(x, SqlParserPos.ZERO))
           .collect(Collectors.toList());
-      for (SqlIdentifier fieldName : fieldNames) {
+      for (SqlIdentifier fieldName : identifiers) {
         w.sep(",");
         fieldName.unparse(w, 0, 0);
       }