[wip] fix: add proto roundtrips for Spark tests and fix issues it surfaces

core/src/main/java/io/substrait/expression/proto/ProtoExpressionConverter.java

@@ -15,6 +15,7 @@
 import io.substrait.relation.ConsistentPartitionWindow;
 import io.substrait.relation.ProtoRelConverter;
 import io.substrait.type.Type;
+import io.substrait.type.TypeVisitor;
 import io.substrait.type.proto.ProtoTypeConverter;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -196,7 +197,20 @@ public Expression from(io.substrait.proto.Expression expr) {
             var rel = protoRelConverter.from(expr.getSubquery().getScalar().getInput());
             yield ImmutableExpression.ScalarSubquery.builder()
                 .input(rel)
-                .type(rel.getRecordType())
+                .type(
+                    rel.getRecordType()
+                        .accept(
+                            new TypeVisitor.TypeThrowsVisitor<Type, RuntimeException>(
+                                "Expected struct field") {
+                              @Override
+                              public Type visit(Type.Struct type) throws RuntimeException {
+                                if (type.fields().size() != 1) {
+                                  throw new UnsupportedOperationException(
+                                      "Scalar subquery must have exactly one field");
+                                }
+                                return type.fields().get(0);
+                              }
+                            }))
                 .build();
           }
           case IN_PREDICATE -> {

core/src/main/java/io/substrait/relation/Join.java

@@ -59,6 +59,7 @@ protected Type.Struct deriveRecordType() {
         switch (getJoinType()) {
           case LEFT, OUTER -> getRight().getRecordType().fields().stream()
               .map(TypeCreator::asNullable);
+          case SEMI, ANTI -> Stream.of(); // these are left joins which ignore right side columns
           default -> getRight().getRecordType().fields().stream();
         };
     return TypeCreator.REQUIRED.struct(Stream.concat(leftTypes, rightTypes));

core/src/test/java/io/substrait/type/proto/ExtensionRoundtripTest.java

@@ -312,7 +312,7 @@ void scalarSubquery() {
                   Stream.of(
                           Expression.ScalarSubquery.builder()
                               .input(relWithEnhancement)
-                              .type(TypeCreator.REQUIRED.struct(TypeCreator.REQUIRED.I64))
+                              .type(TypeCreator.REQUIRED.I64)
                               .build())
                       .collect(Collectors.toList()),
               commonTable);

spark/src/main/scala/io/substrait/debug/RelToVerboseString.scala

@@ -100,6 +100,7 @@ class RelToVerboseString(addSuffix: Boolean) extends DefaultRelVisitor[String] {
         builder.append("commonExtension=").append(commonExtension)
       })
   }
+
   override def visit(namedScan: NamedScan): String = {
     withBuilder(namedScan, 10)(
       builder => {
@@ -115,6 +116,21 @@ class RelToVerboseString(addSuffix: Boolean) extends DefaultRelVisitor[String] {
       })
   }
 
+  override def visit(virtualTableScan: VirtualTableScan): String = {
+    withBuilder(virtualTableScan, 10)(
+      builder => {
+        fillReadRel(virtualTableScan, builder)
+        builder.append(", ")
+        builder.append("rows=").append(virtualTableScan.getRows)
+
+        virtualTableScan.getExtension.ifPresent(
+          extension => {
+            builder.append(", ")
+            builder.append("extension=").append(extension)
+          })
+      })
+  }
+
   override def visit(emptyScan: EmptyScan): String = {
     withBuilder(emptyScan, 10)(
       builder => {

spark/src/main/scala/io/substrait/spark/logical/ToSubstraitRel.scala

@@ -131,7 +131,7 @@ class ToSubstraitRel extends AbstractLogicalPlanVisitor with Logging {
     val aggregates = collectAggregates(actualResultExprs, aggExprToOutputOrdinal)
     val aggOutputMap = aggregates.zipWithIndex.map {
       case (e, i) =>
-        AttributeReference(s"agg_func_$i", e.dataType)() -> e
+        AttributeReference(s"agg_func_$i", e.dataType, nullable = e.nullable)() -> e
     }
     val aggOutput = aggOutputMap.map(_._1)
 
@@ -145,7 +145,7 @@ class ToSubstraitRel extends AbstractLogicalPlanVisitor with Logging {
     }
     val groupOutputMap = actualGroupExprs.zipWithIndex.map {
       case (e, i) =>
-        AttributeReference(s"group_col_$i", e.dataType)() -> e
+        AttributeReference(s"group_col_$i", e.dataType, nullable = e.nullable)() -> e
     }
     val groupOutput = groupOutputMap.map(_._1)
 
@@ -185,13 +185,10 @@ class ToSubstraitRel extends AbstractLogicalPlanVisitor with Logging {
   }
 
   override def visitWindow(window: Window): relation.Rel = {
-    val windowExpressions = window.windowExpressions.map {
-      case w: WindowExpression => fromWindowCall(w, window.child.output)
-      case a: Alias if a.child.isInstanceOf[WindowExpression] =>
-        fromWindowCall(a.child.asInstanceOf[WindowExpression], window.child.output)
-      case other =>
-        throw new UnsupportedOperationException(s"Unsupported window expression: $other")
-    }.asJava
+    val windowExpressions = window.windowExpressions
+      .flatMap(expr => expr.collect { case w: WindowExpression => w })
+      .map(fromWindowCall(_, window.child.output))
+      .asJava
 
     val partitionExpressions = window.partitionSpec.map(toExpression(window.child.output)).asJava
     val sorts = window.orderSpec.map(toSortField(window.child.output)).asJava