Implement SQL on FHIR views

.editorconfig

@@ -900,7 +900,7 @@ ij_json_keep_indents_on_empty_lines = false
 ij_json_keep_line_breaks = true
 ij_json_space_after_colon = true
 ij_json_space_after_comma = true
-ij_json_space_before_colon = true
+ij_json_space_before_colon = false
 ij_json_space_before_comma = false
 ij_json_spaces_within_braces = true
 ij_json_spaces_within_brackets = false

.gitignore

@@ -29,3 +29,4 @@ _version.py
 metastore_db
 derby.log
 spark-warehouse
+.*.crc

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "sql-on-fhir"]
+	path = sql-on-fhir
+	url = https://github.com/FHIR/sql-on-fhir-v2.git

encoders/src/main/java/au/csiro/pathling/config/EncodingConfiguration.java

@@ -26,6 +26,7 @@
 import java.util.Set;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotNull;
+import au.csiro.pathling.encoders.FhirEncoders;
 import lombok.Builder;
 import lombok.Data;
 
@@ -59,19 +60,5 @@ public class EncodingConfiguration {
    */
   @NotNull
   @Builder.Default
-  private Set<String> openTypes = Set.of(
-      "boolean",
-      "code",
-      "date",
-      "dateTime",
-      "decimal",
-      "integer",
-      "string",
-      "Coding",
-      "CodeableConcept",
-      "Address",
-      "Identifier",
-      "Reference"
-  );
-
+  private Set<String> openTypes = FhirEncoders.STANDARD_OPEN_TYPES;
 }

encoders/src/main/java/au/csiro/pathling/encoders/FhirEncoders.java

@@ -41,6 +41,25 @@
  */
 public class FhirEncoders {
 
+  /**
+   * The reasonable default set of open types to encode with extension values.
+   */
+  public static final Set<String> STANDARD_OPEN_TYPES = Set.of(
+      "boolean",
+      "code",
+      "date",
+      "dateTime",
+      "decimal",
+      "integer",
+      "string",
+      "Coding",
+      "CodeableConcept",
+      "Address",
+      "Identifier",
+      "Reference"
+  );
+
+
   /**
    * Cache of Encoders instances.
    */
@@ -308,6 +327,15 @@ public Builder withOpenTypes(final Set<String> openTypes) {
       return this;
     }
 
+    /**
+     * Sets the reasonable default list of types to be encoded for open types, such as extensions.
+     *
+     * @return this builder
+     */
+    public Builder withStandardOenTypes() {
+      return withOpenTypes(STANDARD_OPEN_TYPES);
+    }
+
     /**
      * Switches on/off the support for extensions in encoders.
      *

encoders/src/main/scala/au/csiro/pathling/encoders/Expressions.scala

@@ -29,7 +29,9 @@
 
 package au.csiro.pathling.encoders
 
+import org.apache.spark.sql.{Column, functions}
 import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedException, UnresolvedExtractValue}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodeGenerator, CodegenContext, ExprCode, FalseLiteral}
@@ -304,3 +306,87 @@ case class AttachExtensions(targetObject: Expression,
     AttachExtensions(newChildren.head, newChildren.tail.head)
   }
 }
+
+
+case class UnresolvedIfArray(value: Expression, arrayExpressions: Expression => Expression, elseExpression: Expression => Expression)
+  extends Expression with Unevaluable with NonSQLExpression {
+
+  override def mapChildren(f: Expression => Expression): Expression = {
+
+    val newValue = f(value)
+
+    if (newValue.resolved) {
+      newValue.dataType match {
+        case ArrayType(_, _) => f(arrayExpressions(newValue))
+        case _ => f(elseExpression(newValue))
+      }
+    }
+    else {
+      copy(value = newValue)
+    }
+  }
+
+  override def dataType: DataType = throw new UnresolvedException("dataType")
+
+  override def nullable: Boolean = throw new UnresolvedException("nullable")
+
+  override lazy val resolved = false
+
+  override def toString: String = s"$value"
+
+  override def children: Seq[Expression] = value :: Nil
+
+  override def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = {
+    UnresolvedIfArray(newChildren.head, arrayExpressions, elseExpression)
+  }
+}
+
+
+case class UnresolvedUnnest(value: Expression)
+  extends Expression with Unevaluable with NonSQLExpression {
+
+  override def mapChildren(f: Expression => Expression): Expression = {
+
+    val newValue = f(value)
+
+    if (newValue.resolved) {
+      newValue.dataType match {
+        case ArrayType(ArrayType(_, _), _) => Flatten(newValue)
+        case _ => newValue
+      }
+    }
+    else {
+      copy(value = newValue)
+    }
+  }
+
+  override def dataType: DataType = throw new UnresolvedException("dataType")
+
+  override def nullable: Boolean = throw new UnresolvedException("nullable")
+
+  override lazy val resolved = false
+
+  override def toString: String = s"$value"
+
+  override def children: Seq[Expression] = value :: Nil
+
+  override def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = {
+    UnresolvedUnnest(newChildren.head)
+  }
+}
+
+
+object ValueFunctions {
+  def ifArray(value: Column, arrayExpressions: Column => Column, elseExpression: Column => Column): Column = {
+    val expr = UnresolvedIfArray(value.alias(value.hashCode().toString).expr,
+      e => arrayExpressions(new Column(e)).expr, e => elseExpression(new Column(e)).expr)
+    new Column(expr)
+  }
+
+  def unnest(value: Column): Column = {
+    val expr = UnresolvedUnnest(value.alias(value.hashCode().toString).expr)
+    new Column(expr)
+  }
+
+}
+

encoders/src/test/java/au/csiro/pathling/encoders/ExpressionsTest.java

@@ -0,0 +1,102 @@
+/*
+ * This is a modified version of the Bunsen library, originally published at
+ * https://github.com/cerner/bunsen.
+ *
+ * Bunsen is copyright 2017 Cerner Innovation, Inc., and is licensed under
+ * the Apache License, version 2.0 (http://www.apache.org/licenses/LICENSE-2.0).
+ *
+ * These modifications are copyright 2023 Commonwealth Scientific and Industrial Research
+ * Organisation (CSIRO) ABN 41 687 119 230.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package au.csiro.pathling.encoders;
+
+import org.apache.spark.sql.*;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static au.csiro.pathling.encoders.ValueFunctions.*;
+
+/**
+ * Test for FHIR encoders.
+ */
+public class ExpressionsTest {
+
+
+    private static SparkSession spark;
+
+    /**
+     * Set up Spark.
+     */
+    @BeforeAll
+    public static void setUp() {
+        spark = SparkSession.builder()
+                .master("local[*]")
+                .appName("testing")
+                .config("spark.driver.bindAddress", "localhost")
+                .config("spark.driver.host", "localhost")
+                .getOrCreate();
+
+
+    }
+
+    /**
+     * Tear down Spark.
+     */
+    @AfterAll
+    public static void tearDown() {
+        spark.stop();
+    }
+
+    public static Column size(Column arr) {
+        return ifArray(arr, functions::size, x -> functions.when(arr.isNotNull(), functions.lit(1)).otherwise(functions.lit(0)));
+    }
+
+    @Test
+    public void testIfArray() {
+        Dataset<Row> ds = spark.range(2).toDF();
+        Dataset<Row> resultDs = ds
+                .withColumn("id_array", functions.array(functions.col("id"), functions.lit(22)))
+                .withColumn("test_single", ifArray(ds.col("id"), x -> functions.array(functions.lit(20)), x -> functions.lit(10)))
+                .withColumn("test_array", ifArray(functions.col("id_array"), x -> functions.array(functions.lit(20)), x -> functions.lit(10)))
+                .withColumn("test_array_with_unresolved", ifArray(functions.col("id_array"), x -> functions.array(functions.col("id_array")), x -> functions.lit(10)))
+                .withColumn("test_lit", ifArray(functions.lit("a1"), x -> functions.array(functions.lit(20)), x -> functions.lit(10)))
+                .withColumn("test_array_lit", ifArray(functions.array(functions.lit("a1")), x -> functions.array(ds.col("id")), x -> ds.col("id")))
+                .withColumn("test_array_lit_lambda", ifArray(functions.filter(functions.array(functions.lit("a1")), c -> c.equalTo("a1")), x -> functions.array(functions.lit(20)), x -> functions.lit(10)))
+                .withColumn("test_array_size", size(functions.col("id_array")))
+                .withColumn("test_array_where", ifArray(functions.col("id_array"), x -> functions.filter(x, c -> size(c).equalTo(functions.col("id"))), x -> functions.lit(0)));
+        resultDs.show();
+        resultDs.collectAsList().forEach(System.out::println);
+    }
+
+    @Test
+    public void testFlatten() {
+        Dataset<Row> ds = spark.range(2).toDF();
+        Dataset<Row> resultDs = ds
+                .withColumn("id_array", functions.array(functions.col("id"), functions.lit(22)))
+                .withColumn("id_array_of_arrays", functions.array(functions.array(functions.col("id"), functions.lit(22)),
+                        functions.array(functions.col("id"), functions.lit(33))))
+                .withColumn("test_unnest_single", unnest(ds.col("id")))
+                .withColumn("test_unnest_array", unnest(functions.col("id_array")))
+                .withColumn("test_unnest_array_of_arrays", unnest(functions.col("id_array_of_arrays")))
+                .withColumn("test_unnest_array_of_arrays_if_array", ifArray(functions.col("id_array"), x -> unnest(functions.col("id_array_of_arrays")), x -> x))
+                .withColumn("test_unnest_array_of_arrays_if_id", ifArray(functions.col("id"), x -> unnest(functions.col("id_array_of_arrays")), x -> x));
+
+        resultDs.show();
+        resultDs.collectAsList().forEach(System.out::println);
+    }
+
+}

fhir-server/src/main/java/au/csiro/pathling/aggregate/AggregateExecutor.java

@@ -17,21 +17,22 @@
 
 package au.csiro.pathling.aggregate;
 
+import static java.util.stream.Collectors.toList;
+
 import au.csiro.pathling.config.QueryConfiguration;
-import au.csiro.pathling.fhirpath.FhirPath;
 import au.csiro.pathling.fhirpath.Materializable;
+import au.csiro.pathling.fhirpath.collection.Collection;
 import au.csiro.pathling.io.Database;
 import au.csiro.pathling.io.source.DataSource;
 import au.csiro.pathling.terminology.TerminologyServiceFactory;
 import ca.uhn.fhir.context.FhirContext;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 import javax.annotation.Nonnull;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.SparkSession;
 import org.hl7.fhir.r4.model.Type;
@@ -73,6 +74,8 @@ public AggregateExecutor(@Nonnull final QueryConfiguration configuration,
   public AggregateResponse execute(@Nonnull final AggregateRequest query) {
     final ResultWithExpressions resultWithExpressions = buildQuery(
         query);
+    resultWithExpressions.getDataset().explain();
+    resultWithExpressions.getDataset().show(1_000, false);
 
     // Translate the result into a response object to be passed back to the user.
     return buildResponse(resultWithExpressions);
@@ -95,16 +98,16 @@ private AggregateResponse buildResponse(
         .map(mapRowToGrouping(resultWithExpressions.getParsedAggregations(),
             resultWithExpressions.getParsedGroupings(),
             resultWithExpressions.getParsedFilters()))
-        .collect(Collectors.toList());
+        .collect(toList());
 
     return new AggregateResponse(groupings);
   }
 
   @Nonnull
   @SuppressWarnings("unchecked")
   private Function<Row, AggregateResponse.Grouping> mapRowToGrouping(
-      @Nonnull final List<FhirPath> aggregations, @Nonnull final List<FhirPath> groupings,
-      @Nonnull final Collection<FhirPath> filters) {
+      @Nonnull final List<Collection> aggregations, @Nonnull final List<Collection> groupings,
+      @Nonnull final java.util.Collection<Collection> filters) {
     return row -> {
       final List<Optional<Type>> labels = new ArrayList<>();
       final List<Optional<Type>> results = new ArrayList<>();
@@ -113,7 +116,7 @@ private Function<Row, AggregateResponse.Grouping> mapRowToGrouping(
         final Materializable<Type> grouping = (Materializable<Type>) groupings.get(i);
         // Delegate to the `getValueFromRow` method within each Materializable path class to extract 
         // the Type value from the Row in the appropriate way.
-        final Optional<Type> label = grouping.getValueFromRow(row, i);
+        final Optional<Type> label = grouping.getFhirValueFromRow(row, i);
         labels.add(label);
       }
 
@@ -122,7 +125,7 @@ private Function<Row, AggregateResponse.Grouping> mapRowToGrouping(
         final Materializable aggregation = (Materializable<Type>) aggregations.get(i);
         // Delegate to the `getValueFromRow` method within each Materializable path class to extract 
         // the Type value from the Row in the appropriate way.
-        final Optional<Type> result = aggregation.getValueFromRow(row, i + groupings.size());
+        final Optional<Type> result = aggregation.getFhirValueFromRow(row, i + groupings.size());
         results.add(result);
       }