Filter out deleted TABLE_STORAGE_METRICS (#947)

Author: misolt

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/snowflake/SnowflakeMetadataConnector.java

@@ -282,7 +282,25 @@ public final void addTasksTo(
 
     if (isAssessment) {
       for (AssessmentQuery item : planner.generateAssessmentQueries()) {
-        addAssessmentQuery(item, out, arguments, ACCOUNT_USAGE_SCHEMA_NAME);
+        String formatString =
+            item.needsOverride
+                ? getOverrideableQuery(arguments, item.formatString, TABLE_STORAGE_METRICS)
+                : item.formatString;
+        String whereCondition;
+        // Check whether the overrides changed anything.
+        if (formatString.equals(item.formatString)) {
+          // Overrides either not applied or equal to default values.
+          whereCondition =
+              " WHERE deleted = FALSE AND schema_dropped IS NULL AND table_dropped IS NULL";
+        } else {
+          whereCondition = "";
+        }
+        // The condition is always passed to String.format. SHOW queries simply ignore it.
+        String query = String.format(formatString, ACCOUNT_USAGE_SCHEMA_NAME, whereCondition);
+        Task<?> task =
+            new JdbcSelectTask(item.zipEntryName, query, TaskCategory.REQUIRED, TaskOptions.DEFAULT)
+                .withHeaderTransformer(item.transformer());
+        out.add(task);
       }
     } else {
       if (!arguments.getDatabases().isEmpty()) {
@@ -303,19 +321,15 @@ public final void addTasksTo(
         }
       } else {
         addAssessmentQuery(
-            SnowflakePlanner.SHOW_EXTERNAL_TABLES, out, arguments, ACCOUNT_USAGE_SCHEMA_NAME);
+            SnowflakePlanner.SHOW_EXTERNAL_TABLES,
+            out,
+            arguments,
+            ACCOUNT_USAGE_SCHEMA_NAME,
+            TaskOptions.DEFAULT);
       }
     }
   }
 
-  private void addAssessmentQuery(
-      @Nonnull AssessmentQuery item,
-      @Nonnull List<? super Task<?>> out,
-      @Nonnull ConnectorArguments arguments,
-      @Nonnull String AU) {
-    addAssessmentQuery(item, out, arguments, AU, TaskOptions.DEFAULT);
-  }
-
   private void addAssessmentQuery(
       @Nonnull AssessmentQuery item,
       @Nonnull List<? super Task<?>> out,

dumper/app/src/main/java/com/google/edwmigration/dumper/application/dumper/connector/snowflake/SnowflakePlanner.java

@@ -200,8 +200,7 @@ Task<?> storageMetricsLiteTask() {
             "comment",
             "deleted");
     String view = "SNOWFLAKE.ACCOUNT_USAGE.TABLE_STORAGE_METRICS";
-    String predicate =
-        "deleted = FALSE" + " AND " + "schema_dropped IS NULL" + " AND " + "table_dropped IS NULL";
+    String predicate = "deleted = FALSE AND schema_dropped IS NULL AND table_dropped IS NULL";
     String query = String.format("SELECT %s FROM %s WHERE %s", selectList, view, predicate);
     ImmutableList<String> header =
         ImmutableList.of(