fix: Fix SQL building for dereference queries (facebookincubator#12989)

Summary:
Pull Request resolved: facebookincubator#12989

Currently the Presto expression fuzzer is broken for dereference queries and we have subsequently removed the flag `--enable_dereference` from continuous runs to stabilize the fuzzer. This issue comes from how test Presto SQL is constructed, namely two particular issues:
1. How field names are attached to their parent
2. How rows with particular field names are constructed (and then dereferenced)
 ---
For the former (point #1), the fuzzer is attempting to execute the following expression:
```
if("c0","c1")["row_field0"]
```
This expression aims to access `row_field0` of row `c1` if `c0` is true. Unfortunately the SQL isn't properly constructed and fails because `row_field0` is undefined:
```
Execute presto sql: SELECT row_field0 as p0, row_number as p1 FROM (t_values)
```
This review updates the logic in both PrestoSql (toCallInputsSql function) and the visit plan node function of the PrestoQuery runner to add the missing SQL.

 ---

For the latter (point #2), the fuzzer attempts to build a row and immediately access a particular field, as an example:
```
CONCAT(c0)[row_field0]
```
or, in essence
```
row(c0).row_field0
```
The problem is two-fold. Firstly, the default field name is of the form `field0`, `field1`, etc., so we will need to update the field name. Secondly, the above is not possible in Presto; it will fail:
```
ROW('hello', 5).field0
```
The way to make the above work is to cast and explicitly define the field names and types:
```
select cast(row('hello', 15) as row(field0 varchar, field1 integer)).field0
```
This PR adds this functionality (see test plan for example).

Reviewed By: kagamiori

Differential Revision: D72755054

fbshipit-source-id: d619eeec0c0f320c476c0056b9e00d16de3c1485

Author: peterenescu

velox/exec/fuzzer/PrestoQueryRunner.cpp

@@ -231,7 +231,12 @@ class PrestoQueryRunnerToSqlPlanNodeVisitor : public PrestoSqlPlanNodeVisitor {
       if (auto field =
               std::dynamic_pointer_cast<const core::FieldAccessTypedExpr>(
                   projection)) {
-        sql << field->name();
+        if (field->isInputColumn()) {
+          sql << field->name();
+        } else {
+          toCallInputsSql(field->inputs(), sql);
+          sql << fmt::format(".{}", field->name());
+        }
       } else if (
           auto call = std::dynamic_pointer_cast<const core::CallTypedExpr>(
               projection)) {

velox/exec/fuzzer/PrestoSql.cpp

@@ -133,7 +133,12 @@ void toCallInputsSql(
     if (auto field =
             std::dynamic_pointer_cast<const core::FieldAccessTypedExpr>(
                 input)) {
-      sql << field->name();
+      if (field->isInputColumn()) {
+        sql << field->name();
+      } else {
+        toCallInputsSql(field->inputs(), sql);
+        sql << fmt::format(".{}", field->name());
+      }
     } else if (
         auto call =
             std::dynamic_pointer_cast<const core::CallTypedExpr>(input)) {
@@ -325,11 +330,10 @@ std::string toCastSql(const core::CastTypedExpr& cast) {
 }
 
 std::string toConcatSql(const core::ConcatTypedExpr& concat) {
-  std::stringstream sql;
-  sql << "row(";
-  toCallInputsSql(concat.inputs(), sql);
-  sql << ")";
-  return sql.str();
+  std::stringstream input;
+  toCallInputsSql(concat.inputs(), input);
+  return fmt::format(
+      "cast(row({}) as {})", input.str(), toTypeSql(concat.type()));
 }
 
 template <typename T>

velox/exec/fuzzer/tests/PrestoSqlTest.cpp

@@ -323,5 +323,28 @@ TEST(PrestoSqlTest, toCallSql) {
       "infinity()");
 }
 
+TEST(PrestoSqlTest, toConcatSql) {
+  auto expression = core::ConcatTypedExpr(
+      {"field0", "field1"},
+      std::vector<core::TypedExprPtr>{
+          std::make_shared<core::ConstantTypedExpr>(VARCHAR(), "a"),
+          std::make_shared<core::ConstantTypedExpr>(VARCHAR(), "b")});
+
+  EXPECT_EQ(
+      toConcatSql(expression),
+      "cast(row('a', 'b') as ROW(field0 VARCHAR, field1 VARCHAR))");
+}
+
+TEST(PrestoSqlTest, toCallInputsSql) {
+  std::stringstream sql;
+  auto expression = std::make_shared<core::FieldAccessTypedExpr>(
+      VARCHAR(),
+      std::make_shared<core::FieldAccessTypedExpr>(VARCHAR(), "c0"),
+      "field0");
+
+  toCallInputsSql({expression}, sql);
+  EXPECT_EQ(sql.str(), "c0.field0");
+}
+
 } // namespace
 } // namespace facebook::velox::exec::test

velox/exec/tests/PlanBuilderTest.cpp

@@ -15,9 +15,12 @@
  */
 #include "velox/exec/tests/utils/PlanBuilder.h"
 #include "velox/common/base/tests/GTestUtils.h"
+#include "velox/core/Expressions.h"
 #include "velox/exec/WindowFunction.h"
 #include "velox/functions/prestosql/aggregates/RegisterAggregateFunctions.h"
 #include "velox/functions/prestosql/registration/RegistrationFunctions.h"
+#include "velox/parse/Expressions.h"
+#include "velox/parse/IExpr.h"
 #include "velox/parse/TypeResolver.h"
 #include "velox/vector/tests/utils/VectorTestBase.h"
 
@@ -250,4 +253,52 @@ TEST_F(PlanBuilderTest, missingOutputType) {
       "outputType must be specified");
 }
 
+TEST_F(PlanBuilderTest, projectExpressions) {
+  // Non-typed Expressions.
+  // Simple field access.
+  auto data = ROW({"c0"}, {BIGINT()});
+  VELOX_CHECK_EQ(
+      PlanBuilder()
+          .tableScan("tmp", data)
+          .projectExpressions(
+              {std::make_shared<core::FieldAccessExpr>("c0", std::nullopt)})
+          .planNode()
+          ->toString(true, false),
+      "-- Project[1][expressions: (c0:BIGINT, ROW[\"c0\"])] -> c0:BIGINT\n");
+  // Dereference test using field access query.
+  data = ROW({"c0"}, {ROW({"field0"}, {BIGINT()})});
+  VELOX_CHECK_EQ(
+      PlanBuilder()
+          .tableScan("tmp", data)
+          .projectExpressions({std::make_shared<core::FieldAccessExpr>(
+              "field0",
+              std::nullopt,
+              std::vector<core::ExprPtr>{
+                  std::make_shared<core::FieldAccessExpr>(
+                      "c0", std::nullopt)})})
+          .planNode()
+          ->toString(true, false),
+      "-- Project[1][expressions: (field0:BIGINT, ROW[\"c0\"][field0])] -> field0:BIGINT\n");
+
+  // Test Typed Expressions
+  VELOX_CHECK_EQ(
+      PlanBuilder()
+          .tableScan("tmp", ROW({"c0"}, {ROW({VARCHAR()})}))
+          .projectExpressions(
+              {std::make_shared<core::FieldAccessTypedExpr>(VARCHAR(), "c0")})
+          .planNode()
+          ->toString(true, false),
+      "-- Project[1][expressions: (p0:VARCHAR, \"c0\")] -> p0:VARCHAR\n");
+  VELOX_CHECK_EQ(
+      PlanBuilder()
+          .tableScan("tmp", ROW({"c0"}, {ROW({VARCHAR()})}))
+          .projectExpressions({std::make_shared<core::FieldAccessTypedExpr>(
+              VARCHAR(),
+              std::make_shared<core::FieldAccessTypedExpr>(VARCHAR(), "c0"),
+              "field0")})
+          .planNode()
+          ->toString(true, false),
+      "-- Project[1][expressions: (p0:VARCHAR, \"c0\"[\"field0\"])] -> p0:VARCHAR\n");
+}
+
 } // namespace facebook::velox::exec::test

velox/exec/tests/PrestoQueryRunnerTest.cpp

@@ -254,6 +254,19 @@ TEST_F(PrestoQueryRunnerTest, toSql) {
         queryRunner->toSql(plan),
         "SELECT avg(c0) filter (where c2) as a0, avg(c1) as a1 FROM tmp");
   }
+
+  // Test dereference queries.
+  {
+    auto plan =
+        PlanBuilder()
+            .tableScan("tmp", ROW({"c0"}, {ROW({"field0"}, {BIGINT()})}))
+            .projectExpressions({std::make_shared<core::FieldAccessTypedExpr>(
+                VARCHAR(),
+                std::make_shared<core::FieldAccessTypedExpr>(VARCHAR(), "c0"),
+                "field0")})
+            .planNode();
+    EXPECT_EQ(queryRunner->toSql(plan), "SELECT c0.field0 as p0 FROM (tmp)");
+  }
 }
 
 TEST_F(PrestoQueryRunnerTest, toSqlJoins) {

velox/exec/tests/utils/PlanBuilder.cpp

@@ -419,6 +419,27 @@ PlanBuilder& PlanBuilder::projectExpressions(
   return *this;
 }
 
+PlanBuilder& PlanBuilder::projectExpressions(
+    const std::vector<std::shared_ptr<const core::ITypedExpr>>& projections) {
+  std::vector<core::TypedExprPtr> expressions;
+  std::vector<std::string> projectNames;
+  for (auto i = 0; i < projections.size(); ++i) {
+    expressions.push_back(projections[i]);
+    if (auto fieldExpr =
+            dynamic_cast<const core::FieldAccessExpr*>(projections[i].get())) {
+      projectNames.push_back(fieldExpr->getFieldName());
+    } else {
+      projectNames.push_back(fmt::format("p{}", i));
+    }
+  }
+  planNode_ = std::make_shared<core::ProjectNode>(
+      nextPlanNodeId(),
+      std::move(projectNames),
+      std::move(expressions),
+      planNode_);
+  return *this;
+}
+
 PlanBuilder& PlanBuilder::project(const std::vector<std::string>& projections) {
   VELOX_CHECK_NOT_NULL(planNode_, "Project cannot be the source node");
   std::vector<std::shared_ptr<const core::IExpr>> expressions;

velox/exec/tests/utils/PlanBuilder.h

@@ -552,6 +552,8 @@ class PlanBuilder {
   /// the type.
   PlanBuilder& projectExpressions(
       const std::vector<core::ExprPtr>& projections);
+  PlanBuilder& projectExpressions(
+      const std::vector<core::TypedExprPtr>& projections);
 
   /// Similar to project() except 'optionalProjections' could be empty and the
   /// function will skip creating a ProjectNode in that case.