Only compare recent changes in Datastore and SQL (#1485)

* Only compare recent changes in Datastore and SQL

When comparing Datastore and SQL, ignore older History and EPP resource
objects. This cuts the run time in half compared with a full comparison.
The intention is to run a full comparison before the switch-over from
Datastore and SQL, and run this incremental comparison during the down
time.

The incremental comparison takes about 25 minutes in production.
Performance can be improved further by filtering out older billing
events (OneTime and Cancellation). However, we don't think further
optimization is worth the effort (considering that Recurring events
cannot be filtered since they are mutable but without lastUpdateTime).

Verified in Sandbox and prod with and without time filter.

core/src/test/java/google/registry/beam/comparedb/DatastoreSnapshotsTest.java

@@ -28,6 +28,7 @@ import google.registry.testing.FakeClock;
 import google.registry.testing.InjectExtension;
 import java.io.Serializable;
 import java.nio.file.Path;
+import java.util.Optional;
 import org.apache.beam.sdk.coders.SerializableCoder;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.values.PCollectionTuple;
@@ -83,7 +84,8 @@ class DatastoreSnapshotsTest {
             setupHelper.commitLogDir.getAbsolutePath(),
             START_TIME,
             fakeClock.nowUtc().plusMillis(1),
-            ImmutableSet.copyOf(DatastoreSetupHelper.ALL_KINDS));
+            ImmutableSet.copyOf(DatastoreSetupHelper.ALL_KINDS),
+            Optional.empty());
     PAssert.that(tuple.get(ValidateSqlUtils.createSqlEntityTupleTag(Registrar.class)))
         .containsInAnyOrder(setupHelper.registrar1, setupHelper.registrar2);
     PAssert.that(tuple.get(ValidateSqlUtils.createSqlEntityTupleTag(DomainHistory.class)))

core/src/test/java/google/registry/beam/comparedb/SqlSnapshotsTest.java

@@ -88,6 +88,7 @@ class SqlSnapshotsTest {
                 DomainHistory.class,
                 ContactResource.class,
                 HostResource.class),
+            Optional.empty(),
             Optional.empty());
     PAssert.that(sqlSnapshot.get(createSqlEntityTupleTag(Registry.class)))
         .containsInAnyOrder(setupHelper.registry);