Change canonical tables to views (#1189)

Author: bashir2

pipelines/batch/src/main/java/com/google/fhir/analytics/ConvertResourceFn.java

@@ -206,6 +206,8 @@ public void processElement(ProcessContext processContext)
 
   private Resource createNewFhirResource(String fhirVersion, String resourceType) {
     try {
+      // TODO create tests for this method and different versions of FHIR; casting to R4 resource
+      //  does not seem right!
       return (Resource)
           Class.forName(getFhirBasePackageName(fhirVersion) + "." + resourceType)
               .getConstructor()

pipelines/controller/src/main/java/com/google/fhir/analytics/HiveTableManager.java

@@ -96,21 +96,21 @@ private synchronized void createTablesForResource(
       Connection connection, String resource, String timestamp, String thriftServerParquetPath)
       throws SQLException {
 
+    String location =
+        String.format("%s/%s/%s", THRIFT_CONTAINER_PARQUET_DIR, thriftServerParquetPath, resource);
+    String tableName = String.format("%s_%s", resource, timestamp);
     String sql =
         String.format(
-            "CREATE TABLE IF NOT EXISTS default.%s_%s USING PARQUET LOCATION '%s/%s/%s'",
-            resource, timestamp, THRIFT_CONTAINER_PARQUET_DIR, thriftServerParquetPath, resource);
+            "CREATE TABLE IF NOT EXISTS default.%s USING PARQUET LOCATION '%s'",
+            tableName, location);
     executeSql(connection, sql);
 
-    // Drop canonical table if exists.
-    sql = String.format("DROP TABLE IF EXISTS default.%s", resource);
-    executeSql(connection, sql);
-
-    // Create canonical table with latest parquet files.
+    // Instead of DROP and CREATE we use a VIEW for canonical tables such that the update happens
+    // in one statement (because of lack of transactions in Hive JDBC driver). ALTER TABLE has its
+    // own problems too, e.g., it does not seem to trigger parsing/changing schema.
     sql =
         String.format(
-            "CREATE TABLE IF NOT EXISTS default.%s USING PARQUET LOCATION '%s/%s/%s'",
-            resource, THRIFT_CONTAINER_PARQUET_DIR, thriftServerParquetPath, resource);
+            "CREATE OR REPLACE VIEW default.%s AS SELECT * FROM default.%s", resource, tableName);
     executeSql(connection, sql);
   }
 

pipelines/controller/src/main/java/com/google/fhir/analytics/PipelineManager.java

@@ -568,7 +568,7 @@ public void createResourceTables() {
 
       Preconditions.checkState(paths != null, "Make sure DWH prefix is a valid path!");
 
-      // Sort snapshots directories.
+      // Sort snapshots directories such that the canonical view is created for the latest one.
       Collections.sort(paths, Comparator.comparing(ResourceId::toString));
 
       for (ResourceId path : paths) {
@@ -579,15 +579,20 @@ public void createResourceTables() {
           String fileSeparator = DwhFiles.getFileSeparatorForDwhFiles(rootPrefix);
           List<String> existingResources =
               dwhFilesManager.findExistingResources(baseDir + fileSeparator + path.getFilename());
-          hiveTableManager.createResourceAndCanonicalTables(
-              existingResources, timestamp, path.getFilename());
+          try {
+            hiveTableManager.createResourceAndCanonicalTables(
+                existingResources, timestamp, path.getFilename());
+          } catch (SQLException e) {
+            logger.error(
+                "Exception while creating resource table on thriftserver for path: {}",
+                path.getFilename(),
+                e);
+          }
         }
       }
     } catch (IOException e) {
       // In case of exceptions at this stage, we just log the exception.
       logger.error("Exception while reading thriftserver parquet output directory: ", e);
-    } catch (SQLException e) {
-      logger.error("Exception while creating resource tables on thriftserver: ", e);
     }
   }