Run InitSqlPipeline (#727)

* Run InitSqlPipeline

Added the main() method to InitSqlPipeline.

Added a Gradle task to run InitSqlPipeline from command line. This
task is meant for testing and experiments.

Corrected the file name prefix of Datastore export files. Should
be 'output-', defined as 'input-'.

core/build.gradle

@@ -819,6 +819,30 @@ createToolTask(
     'google.registry.tools.DevTool',
     sourceSets.nonprod)
 
+project.tasks.create('initSqlPipeline', JavaExec) {
+  main = 'google.registry.beam.initsql.InitSqlPipeline'
+
+  doFirst {
+    getToolArgsList().ifPresent {
+      args it
+    }
+
+    def isDirectRunner =
+        args.contains('DirectRunner') || args.contains('--runner=DirectRunner')
+    // The dependency containing DirectRunner is intentionally excluded from the
+    // production binary, so that it won't be chosen by mistake: we definitely do
+    // not want to use it for the real jobs, yet DirectRunner is the default if
+    // the user forgets to override it.
+    // DirectRunner is required for tests and is already on testRuntimeClasspath.
+    // For simplicity, we add testRuntimeClasspath to this task's classpath instead
+    // of defining a new configuration just for the DirectRunner dependency.
+    classpath =
+        isDirectRunner
+            ? sourceSets.main.runtimeClasspath.plus(sourceSets.test.runtimeClasspath)
+            : sourceSets.main.runtimeClasspath
+  }
+}
+
 project.tasks.create('generateSqlSchema', JavaExec) {
   classpath = sourceSets.nonprod.runtimeClasspath
   main = 'google.registry.tools.DevTool'

core/src/main/java/google/registry/beam/initsql/BackupPaths.java

@@ -32,7 +32,7 @@ public final class BackupPaths {
   private BackupPaths() {}
 
   private static final String WILDCARD_CHAR = "*";
-  private static final String EXPORT_PATTERN_TEMPLATE = "%s/all_namespaces/kind_%s/input-%s";
+  private static final String EXPORT_PATTERN_TEMPLATE = "%s/all_namespaces/kind_%s/output-%s";
 
   public static final String COMMIT_LOG_NAME_PREFIX = "commit_diff_until_";
   private static final String COMMIT_LOG_PATTERN_TEMPLATE = "%s/" + COMMIT_LOG_NAME_PREFIX + "*";

core/src/main/java/google/registry/beam/initsql/InitSqlPipeline.java

@@ -40,6 +40,7 @@ import java.util.Collection;
 import java.util.Optional;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.SerializableFunction;
 import org.apache.beam.sdk.transforms.Wait;
@@ -234,4 +235,10 @@ public class InitSqlPipeline implements Serializable {
       return entityClasses.stream().map(Key::getKind).collect(ImmutableList.toImmutableList());
     }
   }
+
+  public static void main(String[] args) {
+    InitSqlPipelineOptions options =
+        PipelineOptionsFactory.fromArgs(args).withValidation().as(InitSqlPipelineOptions.class);
+    new InitSqlPipeline(options).run();
+  }
 }

core/src/test/java/google/registry/beam/initsql/BackupTestStoreTest.java

@@ -109,9 +109,9 @@ public class BackupTestStoreTest {
             .map(string -> string.substring(exportFolder.getAbsolutePath().length()))) {
       assertThat(files)
           .containsExactly(
-              "/all_namespaces/kind_Registry/input-0",
+              "/all_namespaces/kind_Registry/output-0",
-              "/all_namespaces/kind_DomainBase/input-0",
+              "/all_namespaces/kind_DomainBase/output-0",
-              "/all_namespaces/kind_ContactResource/input-0");
+              "/all_namespaces/kind_ContactResource/output-0");
     }
   }
 
@@ -132,16 +132,16 @@ public class BackupTestStoreTest {
     String exportRootPath = tempDir.getAbsolutePath();
     File exportFolder = export(exportRootPath, Collections.EMPTY_SET);
     ImmutableList<Object> loadedRegistries =
-        loadExportedEntities(new File(exportFolder, "/all_namespaces/kind_Registry/input-0"));
+        loadExportedEntities(new File(exportFolder, "/all_namespaces/kind_Registry/output-0"));
     assertThat(loadedRegistries).containsExactly(registry);
 
     ImmutableList<Object> loadedDomains =
-        loadExportedEntities(new File(exportFolder, "/all_namespaces/kind_DomainBase/input-0"));
+        loadExportedEntities(new File(exportFolder, "/all_namespaces/kind_DomainBase/output-0"));
     assertThat(loadedDomains).containsExactly(domain);
 
     ImmutableList<Object> loadedContacts =
         loadExportedEntities(
-            new File(exportFolder, "/all_namespaces/kind_ContactResource/input-0"));
+            new File(exportFolder, "/all_namespaces/kind_ContactResource/output-0"));
     assertThat(loadedContacts).containsExactly(contact);
   }
 
@@ -156,7 +156,7 @@ public class BackupTestStoreTest {
         export(
             exportRootPath, ImmutableSet.of(Key.create(getCrossTldKey(), Registry.class, "tld1")));
     ImmutableList<Object> loadedRegistries =
-        loadExportedEntities(new File(exportFolder, "/all_namespaces/kind_Registry/input-0"));
+        loadExportedEntities(new File(exportFolder, "/all_namespaces/kind_Registry/output-0"));
     assertThat(loadedRegistries).containsExactly(newRegistry);
   }
 

core/src/test/java/google/registry/beam/initsql/ExportloadingTransformsTest.java

@@ -119,9 +119,9 @@ class ExportloadingTransformsTest implements Serializable {
 
     ImmutableList<String> expectedPatterns =
         ImmutableList.of(
-            exportDir.getAbsolutePath() + "/all_namespaces/kind_Registry/input-*",
+            exportDir.getAbsolutePath() + "/all_namespaces/kind_Registry/output-*",
-            exportDir.getAbsolutePath() + "/all_namespaces/kind_DomainBase/input-*",
+            exportDir.getAbsolutePath() + "/all_namespaces/kind_DomainBase/output-*",
-            exportDir.getAbsolutePath() + "/all_namespaces/kind_ContactResource/input-*");
+            exportDir.getAbsolutePath() + "/all_namespaces/kind_ContactResource/output-*");
 
     PAssert.that(patterns).containsInAnyOrder(expectedPatterns);
 
@@ -135,10 +135,10 @@ class ExportloadingTransformsTest implements Serializable {
             .apply(
                 "File patterns to metadata",
                 Create.of(
-                        exportDir.getAbsolutePath() + "/all_namespaces/kind_Registry/input-*",
+                        exportDir.getAbsolutePath() + "/all_namespaces/kind_Registry/output-*",
-                        exportDir.getAbsolutePath() + "/all_namespaces/kind_DomainBase/input-*",
+                        exportDir.getAbsolutePath() + "/all_namespaces/kind_DomainBase/output-*",
                         exportDir.getAbsolutePath()
-                            + "/all_namespaces/kind_ContactResource/input-*")
+                            + "/all_namespaces/kind_ContactResource/output-*")
                     .withCoder(StringUtf8Coder.of()))
             .apply(Transforms.getFilesByPatterns());
 
@@ -157,9 +157,9 @@ class ExportloadingTransformsTest implements Serializable {
 
     ImmutableList<String> expectedFilenames =
         ImmutableList.of(
-            exportDir.getAbsolutePath() + "/all_namespaces/kind_Registry/input-0",
+            exportDir.getAbsolutePath() + "/all_namespaces/kind_Registry/output-0",
-            exportDir.getAbsolutePath() + "/all_namespaces/kind_DomainBase/input-0",
+            exportDir.getAbsolutePath() + "/all_namespaces/kind_DomainBase/output-0",
-            exportDir.getAbsolutePath() + "/all_namespaces/kind_ContactResource/input-0");
+            exportDir.getAbsolutePath() + "/all_namespaces/kind_ContactResource/output-0");
 
     PAssert.that(fileNames).containsInAnyOrder(expectedFilenames);