fix(isthmus): elide redundant identity projections

isthmus/src/main/java/io/substrait/isthmus/SubstraitRelVisitor.java

@@ -58,6 +58,7 @@
 import org.apache.calcite.rex.RexFieldAccess;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.util.ImmutableBitSet;
@@ -208,6 +209,15 @@ public Rel visit(org.apache.calcite.rel.core.Calc calc) {
    */
   @Override
   public Rel visit(org.apache.calcite.rel.core.Project project) {
+    // An identity projection (input refs in order, with matching types) passes every input field
+    // through unchanged and only ever renames fields. Substrait carries output names on Plan.Root,
+    // not on the Project, so emitting one here is redundant: the reverse conversion drops it, which
+    // leaves the two sides structurally different and breaks round-trips. Skip it instead. Output
+    // names are still preserved because convert(RelRoot, ...) takes them from validatedRowType.
+    if (RexUtil.isIdentity(project.getProjects(), project.getInput().getRowType())) {
+      return apply(project.getInput());
+    }
+
     List<Expression> expressions =
         project.getProjects().stream()
             .map(this::toExpression)
@@ -218,7 +228,9 @@ public Rel visit(org.apache.calcite.rel.core.Project project) {
       return Project.builder().expressions(expressions).input(apply(project.getInput())).build();
     }
 
-    // todo: eliminate excessive projects. This should be done by converting rexinputrefs to remaps.
+    // todo: eliminate the remaining excessive projects. Identity projects are dropped above; a
+    // projection whose expressions are all input refs (a permutation or column pruning) could
+    // likewise be expressed as a remap over the input rather than copied expressions.
     return Project.builder()
         .remap(
             Rel.Remap.offset(project.getInput().getRowType().getFieldCount(), expressions.size()))

isthmus/src/test/java/io/substrait/isthmus/DdlRoundtripTest.java

@@ -18,12 +18,12 @@ public DdlRoundtripTest() throws SqlParseException {
   @Test
   void testCreateTable() throws Exception {
     String sql = "create table dst1 as select * from src1";
-    assertFullRoundTripWithIdentityProjectionWorkaround(sql, catalogReader);
+    assertFullRoundTrip(sql, catalogReader);
   }
 
   @Test
   void testCreateView() throws Exception {
     String sql = "create view dst1 as select * from src1";
-    assertFullRoundTripWithIdentityProjectionWorkaround(sql, catalogReader);
+    assertFullRoundTrip(sql, catalogReader);
   }
 }

isthmus/src/test/java/io/substrait/isthmus/DmlRoundtripTest.java

@@ -16,21 +16,18 @@ public DmlRoundtripTest() throws SqlParseException {}
 
   @Test
   void testDelete() throws SqlParseException {
-    assertFullRoundTripWithIdentityProjectionWorkaround(
-        "delete from src1 where intcol=10", catalogReader);
+    assertFullRoundTrip("delete from src1 where intcol=10", catalogReader);
   }
 
   @Test
   void testUpdate() throws SqlParseException {
-    assertFullRoundTripWithIdentityProjectionWorkaround(
-        "update src1 set intcol=10 where charcol='a'", catalogReader);
+    assertFullRoundTrip("update src1 set intcol=10 where charcol='a'", catalogReader);
   }
 
   @Test
   void testInsert() throws SqlParseException {
-    assertFullRoundTripWithIdentityProjectionWorkaround(
-        "insert into src1 (intcol, charcol) values (1,'a'); ", catalogReader);
-    assertFullRoundTripWithIdentityProjectionWorkaround(
+    assertFullRoundTrip("insert into src1 (intcol, charcol) values (1,'a'); ", catalogReader);
+    assertFullRoundTrip(
         "insert into src1 (intcol, charcol) select intcol,charcol from src2;", catalogReader);
   }
 }

isthmus/src/test/java/io/substrait/isthmus/NameRoundtripTest.java

@@ -39,6 +39,15 @@ void preserveNamesFromSql() throws Exception {
     assertEquals(expectedNames, calciteRelRoot2.validatedRowType.getFieldNames());
   }
 
+  @Test
+  void roundTripNameChangingProjectionOverAggregate() throws Exception {
+    // Calcite wraps the aggregate in a name-changing identity projection (b -> B). That projection
+    // is redundant in Substrait (output names live on Plan.Root), so it must not break the round
+    // trip by surviving on only one side of the conversion.
+    assertFullRoundTrip(
+        "SELECT \"a\", \"B\" FROM foo GROUP BY a, b", "CREATE TABLE foo(a BIGINT, b BIGINT)");
+  }
+
   @Test
   void preserveNamesFromSubstrait() {
     NamedScan rel =

isthmus/src/test/java/io/substrait/isthmus/PlanTestBase.java

@@ -276,75 +276,6 @@ protected void assertFullRoundTrip(String sqlQuery, Prepare.CatalogReader catalo
     assertEquals(root1, root3);
   }
 
-  /**
-   * Verifies that the given query can be converted from its Calcite representation to Substrait
-   * proto and back. Due to the various ways in which Calcite plans are produced, some plans contain
-   * identity projections and others do not. Fully removing this behaviour is quite tricky. As a
-   * workaround, this method prepares the plan such that identity projections are removed.
-   *
-   * <p>In the long-term, we should work to remove this test method.
-   *
-   * <p>Preparation: <code>
-   *   SQL -> Calcite 0 -> Substrait POJO 0 -> Substrait Proto 0 -> Substrait POJO 1 -> Calcite 1
-   * </code> this code also checks that: Main cycle:
-   *
-   * <ul>
-   *   <li>Substrait POJO 0 == Substrait POJO 1
-   * </ul>
-   *
-   * Calcite 1 -> Substrait POJO 2 -> Substrait Proto 2 -> Substrait POJO 3 -> Calcite 2 ->
-   * Substrait POJO 4
-   *
-   * <ul>
-   *   <li>Substrait POJO 2 == Substrait POJO 4
-   * </ul>
-   */
-  protected void assertFullRoundTripWithIdentityProjectionWorkaround(
-      String sqlQuery, Prepare.CatalogReader catalogReader) throws SqlParseException {
-    ExtensionCollector extensionCollector = new ExtensionCollector();
-
-    // Preparation
-    // SQL -> Calcite 0
-    RelRoot calcite0 = SubstraitSqlToCalcite.convertQuery(sqlQuery, catalogReader);
-
-    // Calcite 0 -> Substrait POJO 0
-    Plan.Root root0 = SubstraitRelVisitor.convert(calcite0, converterProvider);
-
-    // Substrait POJO 0 -> Substrait Proto 0
-    io.substrait.proto.RelRoot proto0 = new RelProtoConverter(extensionCollector).toProto(root0);
-
-    // Substrait Proto -> Substrait POJO 1
-    Plan.Root root1 = new ProtoRelConverter(extensionCollector, extensions).from(proto0);
-
-    // Verify that POJOs are the same
-    assertEquals(root0, root1);
-
-    final SubstraitToCalcite substraitToCalcite =
-        new SubstraitToCalcite(converterProvider, catalogReader);
-
-    // Substrait POJO 1 -> Calcite 1
-    RelRoot calcite1 = substraitToCalcite.convert(root1);
-
-    // End Preparation
-
-    // Calcite 1 -> Substrait POJO 2
-    Plan.Root root2 = SubstraitRelVisitor.convert(calcite1, converterProvider);
-
-    // Substrait POJO 2 -> Substrait Proto 1
-    io.substrait.proto.RelRoot proto1 = new RelProtoConverter(extensionCollector).toProto(root2);
-
-    // Substrait Proto1 -> Substrait POJO 3
-    Plan.Root root3 = new ProtoRelConverter(extensionCollector, extensions).from(proto1);
-
-    // Substrait POJO 3 -> Calcite 2
-    RelRoot calcite2 = substraitToCalcite.convert(root3);
-    // Calcite 2 -> Substrait POJO 4
-    Plan.Root root4 = SubstraitRelVisitor.convert(calcite2, converterProvider);
-
-    // Verify that POJOs are the same
-    assertEquals(root2, root4);
-  }
-
   /**
    * Verifies that the given POJO can be converted:
    *

isthmus/src/test/java/io/substrait/isthmus/SubqueryPlanTest.java

@@ -221,13 +221,7 @@ void existsNestedCorrelatedSubquery() throws IOException, SqlParseException {
     assertSame(PredicateOp.PREDICATE_OP_UNIQUE, inner_subquery.getSetPredicate().getPredicateOp());
 
     Expression inner_subquery_condition =
-        inner_subquery
-            .getSetPredicate()
-            .getTuples()
-            .getProject()
-            .getInput()
-            .getFilter()
-            .getCondition();
+        inner_subquery.getSetPredicate().getTuples().getFilter().getCondition();
 
     Expression inner_subquery_cond1 =
         inner_subquery_condition