Implement input for MRs over child entities

Also throwing in a proof-of-concept MR that I'd like to run in production, and then scrap once the meaty MR is finished (e.g. exploding Recurring billing events into OneTimes).
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=119881471

java/com/google/domain/registry/env/common/tools/WEB-INF/web.xml

@@ -88,6 +88,12 @@
     <url-pattern>/_dr/task/resaveAllEppResources</url-pattern>
   </servlet-mapping>
 
+  <!--  Mapreduce to count recurring billing events (to test the child entity reader). -->
+  <servlet-mapping>
+    <servlet-name>tools-servlet</servlet-name>
+    <url-pattern>/_dr/task/countRecurringBillingEvents</url-pattern>
+  </servlet-mapping>
+
   <!-- This path serves up the App Engine results page for mapreduce runs. -->
   <servlet>
     <servlet-name>mapreduce</servlet-name>

java/com/google/domain/registry/mapreduce/inputs/ChildEntityInput.java

@@ -0,0 +1,53 @@
+// Copyright 2016 The Domain Registry Authors. All Rights Reserved.
+//
+// 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 com.google.domain.registry.mapreduce.inputs;
+
+import static com.google.domain.registry.util.TypeUtils.checkNoInheritanceRelationships;
+
+import com.google.appengine.tools.mapreduce.Input;
+import com.google.appengine.tools.mapreduce.InputReader;
+import com.google.common.collect.ImmutableSet;
+import com.google.domain.registry.model.EppResource;
+import com.google.domain.registry.model.ImmutableObject;
+import com.google.domain.registry.model.index.EppResourceIndexBucket;
+
+import com.googlecode.objectify.Key;
+
+/**
+ * A MapReduce {@link Input} that loads all child objects of a given set of types, that are children
+ * of given {@link EppResource} types.
+ */
+class ChildEntityInput<R extends EppResource, I extends ImmutableObject>
+    extends EppResourceBaseInput<I> {
+
+  private static final long serialVersionUID = -3888034213150865008L;
+
+  private final ImmutableSet<Class<? extends R>> resourceClasses;
+  private final ImmutableSet<Class<? extends I>> childResourceClasses;
+
+  public ChildEntityInput(
+      ImmutableSet<Class<? extends R>> resourceClasses,
+      ImmutableSet<Class<? extends I>> childResourceClasses) {
+    this.resourceClasses = resourceClasses;
+    this.childResourceClasses = childResourceClasses;
+    checkNoInheritanceRelationships(ImmutableSet.<Class<?>>copyOf(resourceClasses));
+    checkNoInheritanceRelationships(ImmutableSet.<Class<?>>copyOf(childResourceClasses));
+  }
+
+  @Override
+  protected InputReader<I> bucketToReader(Key<EppResourceIndexBucket> bucketKey) {
+    return new ChildEntityReader<>(bucketKey, resourceClasses, childResourceClasses);
+  }
+}

java/com/google/domain/registry/mapreduce/inputs/ChildEntityReader.java

@@ -0,0 +1,192 @@
+// Copyright 2016 The Domain Registry Authors. All Rights Reserved.
+//
+// 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 com.google.domain.registry.mapreduce.inputs;
+
+import static com.google.domain.registry.model.EntityClasses.ALL_CLASSES;
+import static com.google.domain.registry.model.ofy.ObjectifyService.ofy;
+
+import com.google.appengine.api.datastore.Cursor;
+import com.google.appengine.api.datastore.QueryResultIterator;
+import com.google.appengine.tools.mapreduce.InputReader;
+import com.google.appengine.tools.mapreduce.ShardContext;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.domain.registry.model.EppResource;
+import com.google.domain.registry.model.ImmutableObject;
+import com.google.domain.registry.model.index.EppResourceIndex;
+import com.google.domain.registry.model.index.EppResourceIndexBucket;
+import com.google.domain.registry.util.FormattingLogger;
+
+import com.googlecode.objectify.Key;
+import com.googlecode.objectify.annotation.Entity;
+import com.googlecode.objectify.cmd.Query;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+
+/**
+ * Reader that maps over {@link EppResourceIndex} and returns resources that are children of
+ * {@link EppResource} objects.
+ */
+class ChildEntityReader<R extends EppResource, I extends ImmutableObject> extends InputReader<I> {
+
+  private static final long serialVersionUID = -7430731417793849164L;
+
+  static final FormattingLogger logger = FormattingLogger.getLoggerForCallerClass();
+
+  /** This reader uses an EppResourceEntityReader under the covers to iterate over EPP resources. */
+  private final EppResourceEntityReader<? extends R> eppResourceEntityReader;
+  /** The current EPP resource being referenced for child entity queries. */
+  private Key<? extends R> currentEppResource;
+
+  /** The child resource classes to postfilter for. */
+  private final ImmutableList<Class<? extends I>> childResourceClasses;
+  /** The index within the list above for the next ofy query. */
+  private int childResourceClassIndex;
+
+  /** An iterator over queries for child entities of EppResources. */
+  private transient QueryResultIterator<I> childQueryIterator;
+  /** A cursor for queries for child entities of EppResources. */
+  private Cursor childCursor;
+
+  public ChildEntityReader(
+      Key<EppResourceIndexBucket> bucketKey,
+      ImmutableSet<Class<? extends R>> resourceClasses,
+      ImmutableSet<Class<? extends I>> childResourceClasses) {
+    this.childResourceClasses = expandPolymorphicClasses(childResourceClasses);
+    this.eppResourceEntityReader = new EppResourceEntityReader<>(bucketKey, resourceClasses);
+  }
+
+  /** Expands non-entity polymorphic classes into their child types. */
+  @SuppressWarnings("unchecked")
+  private ImmutableList<Class<? extends I>> expandPolymorphicClasses(
+      ImmutableSet<Class<? extends I>> resourceClasses) {
+    ImmutableList.Builder<Class<? extends I>> builder = ImmutableList.builder();
+    for (Class<? extends I> clazz : resourceClasses) {
+      if (clazz.isAnnotationPresent(Entity.class)) {
+        builder.add(clazz);
+      } else {
+        for (Class<? extends ImmutableObject> entityClass : ALL_CLASSES) {
+          if (clazz.isAssignableFrom(entityClass)) {
+            builder.add((Class<? extends I>) entityClass);
+          }
+        }
+      }
+    }
+    return builder.build();
+  }
+
+  /**
+   * Get the next {@link ImmutableObject} (i.e. child element) from the query.
+   *
+   * @throws NoSuchElementException if there are no more EPP resources to iterate over.
+   */
+  I nextChild() throws NoSuchElementException {
+    try {
+      while (true) {
+        if (currentEppResource == null) {
+          currentEppResource = Key.create(eppResourceEntityReader.next());
+          childResourceClassIndex = 0;
+          childQueryIterator = null;
+        }
+        if (childQueryIterator == null) {
+          childQueryIterator = childQuery().iterator();
+        }
+        try {
+          return childQueryIterator.next();
+        } catch (NoSuchElementException e) {
+          childQueryIterator = null;
+          childResourceClassIndex++;
+          if (childResourceClassIndex >= childResourceClasses.size()) {
+            currentEppResource = null;
+          }
+        }
+      }
+    } finally {
+      ofy().clearSessionCache(); // Try not to leak memory.
+    }
+  }
+
+  @Override
+  public I next() throws NoSuchElementException {
+    while (true) {
+      I entity = nextChild();
+      if (entity != null) {
+        // Postfilter to distinguish polymorphic types.
+        for (Class<? extends I> resourceClass : childResourceClasses) {
+          if (resourceClass.isInstance(entity)) {
+            return entity;
+          }
+        }
+      }
+    }
+  }
+
+  /** Query for children of the current resource and of the current child class. */
+  private Query<I> childQuery() {
+    @SuppressWarnings("unchecked")
+    Query<I> query = (Query<I>) ofy().load()
+        .type(childResourceClasses.get(childResourceClassIndex))
+        .ancestor(currentEppResource);
+    return query;
+  }
+
+  @Override
+  public void beginSlice() {
+    eppResourceEntityReader.beginSlice();
+    if (childCursor != null) {
+      Query<I> query = childQuery().startAt(childCursor);
+      childQueryIterator = query.iterator();
+    }
+  }
+
+  @Override
+  public void endSlice() {
+    if (childQueryIterator != null) {
+      childCursor = childQueryIterator.getCursor();
+    }
+    eppResourceEntityReader.endSlice();
+  }
+
+  @Override
+  public Double getProgress() {
+    return eppResourceEntityReader.getProgress();
+  }
+
+  @Override
+  public long estimateMemoryRequirement() {
+    return eppResourceEntityReader.estimateMemoryRequirement();
+  }
+
+  @Override
+  public ShardContext getContext() {
+    return eppResourceEntityReader.getContext();
+  }
+
+  @Override
+  public void setContext(ShardContext context) {
+    eppResourceEntityReader.setContext(context);
+  }
+
+  @Override
+  public void beginShard() throws IOException {
+    eppResourceEntityReader.beginShard();
+  }
+
+  @Override
+  public void endShard() throws IOException {
+    eppResourceEntityReader.endShard();
+  }
+}

java/com/google/domain/registry/mapreduce/inputs/EppResourceBaseInput.java

@@ -14,14 +14,9 @@
 
 package com.google.domain.registry.mapreduce.inputs;
 
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.domain.registry.util.CollectionUtils.difference;
-
 import com.google.appengine.tools.mapreduce.Input;
 import com.google.appengine.tools.mapreduce.InputReader;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
-import com.google.domain.registry.model.EppResource;
 import com.google.domain.registry.model.index.EppResourceIndex;
 import com.google.domain.registry.model.index.EppResourceIndexBucket;
 
@@ -45,18 +40,5 @@ abstract class EppResourceBaseInput<I> extends Input<I> {
 
   /** Creates a reader that returns the resources under a bucket. */
   protected abstract InputReader<I> bucketToReader(Key<EppResourceIndexBucket> bucketKey);
-
-  static <R extends EppResource> void checkResourceClassesForInheritance(
-      ImmutableSet<Class<? extends R>> resourceClasses) {
-    for (Class<? extends R> resourceClass : resourceClasses) {
-      for (Class<? extends R> potentialSuperclass : difference(resourceClasses, resourceClass)) {
-        checkArgument(
-            !potentialSuperclass.isAssignableFrom(resourceClass),
-            "Cannot specify resource classes with inheritance relationship: %s extends %s",
-            resourceClass,
-            potentialSuperclass);
-      }
-    }
-  }
 }
 

java/com/google/domain/registry/mapreduce/inputs/EppResourceBaseReader.java

@@ -61,8 +61,7 @@ abstract class EppResourceBaseReader<T> extends InputReader<T> {
   private transient QueryResultIterator<EppResourceIndex> queryIterator;
 
   EppResourceBaseReader(
-      Key<EppResourceIndexBucket>
+      Key<EppResourceIndexBucket> bucketKey,
-      bucketKey,
       long memoryEstimate,
       ImmutableSet<String> filterKinds) {
     this.bucketKey = bucketKey;

java/com/google/domain/registry/mapreduce/inputs/EppResourceEntityInput.java

@@ -14,6 +14,8 @@
 
 package com.google.domain.registry.mapreduce.inputs;
 
+import static com.google.domain.registry.util.TypeUtils.checkNoInheritanceRelationships;
+
 import com.google.appengine.tools.mapreduce.Input;
 import com.google.appengine.tools.mapreduce.InputReader;
 import com.google.common.collect.ImmutableSet;
@@ -31,7 +33,7 @@ class EppResourceEntityInput<R extends EppResource> extends EppResourceBaseInput
 
   public EppResourceEntityInput(ImmutableSet<Class<? extends R>> resourceClasses) {
     this.resourceClasses = resourceClasses;
-    checkResourceClassesForInheritance(resourceClasses);
+    checkNoInheritanceRelationships(ImmutableSet.<Class<?>>copyOf(resourceClasses));
   }
 
   @Override
@@ -39,5 +41,3 @@ class EppResourceEntityInput<R extends EppResource> extends EppResourceBaseInput
     return new EppResourceEntityReader<R>(bucketKey, resourceClasses);
   }
 }
-
-

java/com/google/domain/registry/mapreduce/inputs/EppResourceInputs.java

@@ -23,6 +23,7 @@ import static com.google.domain.registry.util.TypeUtils.hasAnnotation;
 import com.google.appengine.tools.mapreduce.Input;
 import com.google.common.collect.ImmutableSet;
 import com.google.domain.registry.model.EppResource;
+import com.google.domain.registry.model.ImmutableObject;
 import com.google.domain.registry.model.index.EppResourceIndex;
 
 import com.googlecode.objectify.Key;
@@ -59,6 +60,24 @@ public final class EppResourceInputs {
         ImmutableSet.copyOf(asList(resourceClass, moreResourceClasses)));
   }
 
+
+  /**
+   * Returns a MapReduce {@link Input} that loads all {@link ImmutableObject} objects of a given
+   * type, including deleted resources, that are child entities of all {@link EppResource} objects
+   * of a given type.
+   *
+   * <p>Note: Do not concatenate multiple EntityInputs together (this is inefficient as it iterates
+   * through all buckets multiple times). Specify the types in a single input, or load all types by
+   * specifying {@link EppResource} and/or {@link ImmutableObject} as the class.
+   */
+  public static <R extends EppResource, I extends ImmutableObject> Input<I> createChildEntityInput(
+      ImmutableSet<Class<? extends R>> parentClasses,
+      ImmutableSet<Class<? extends I>> childClasses) {
+    checkArgument(!parentClasses.isEmpty(), "Must provide at least one parent type.");
+    checkArgument(!childClasses.isEmpty(), "Must provide at least one child type.");
+    return new ChildEntityInput<>(parentClasses, childClasses);
+  }
+
   /**
    * Returns a MapReduce {@link Input} that loads keys to all {@link EppResource} objects of a given
    * type, including deleted resources.

java/com/google/domain/registry/mapreduce/inputs/EppResourceKeyInput.java

@@ -14,6 +14,8 @@
 
 package com.google.domain.registry.mapreduce.inputs;
 
+import static com.google.domain.registry.util.TypeUtils.checkNoInheritanceRelationships;
+
 import com.google.appengine.tools.mapreduce.Input;
 import com.google.appengine.tools.mapreduce.InputReader;
 import com.google.common.collect.ImmutableSet;
@@ -35,7 +37,7 @@ class EppResourceKeyInput<R extends EppResource> extends EppResourceBaseInput<Ke
 
   public EppResourceKeyInput(ImmutableSet<Class<? extends R>> resourceClasses) {
     this.resourceClasses = resourceClasses;
-    checkResourceClassesForInheritance(resourceClasses);
+    checkNoInheritanceRelationships(ImmutableSet.<Class<?>>copyOf(resourceClasses));
   }
 
   @Override

java/com/google/domain/registry/module/tools/BUILD

@@ -19,6 +19,7 @@ java_library(
         "//java/com/google/domain/registry/request",
         "//java/com/google/domain/registry/request:modules",
         "//java/com/google/domain/registry/tools/server",
+        "//java/com/google/domain/registry/tools/server/javascrap",
         "//java/com/google/domain/registry/util",
         "//third_party/java/bouncycastle",
         "//third_party/java/dagger",

java/com/google/domain/registry/module/tools/ToolsRequestComponent.java

@@ -37,6 +37,7 @@ import com.google.domain.registry.tools.server.ResaveAllEppResourcesAction;
 import com.google.domain.registry.tools.server.ToolsServerModule;
 import com.google.domain.registry.tools.server.UpdatePremiumListAction;
 import com.google.domain.registry.tools.server.VerifyOteAction;
+import com.google.domain.registry.tools.server.javascrap.CountRecurringBillingEventsAction;
 
 import dagger.Subcomponent;
 
@@ -50,6 +51,7 @@ import dagger.Subcomponent;
         ToolsServerModule.class,
     })
 interface ToolsRequestComponent {
+  CountRecurringBillingEventsAction countRecurringBillingEventsAction();
   CreateGroupsAction createGroupsAction();
   CreatePremiumListAction createPremiumListAction();
   DeleteEntityAction deleteEntityAction();

java/com/google/domain/registry/util/TypeUtils.java

@@ -15,11 +15,13 @@
 package com.google.domain.registry.util;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.domain.registry.util.CollectionUtils.difference;
 import static java.lang.reflect.Modifier.isFinal;
 import static java.lang.reflect.Modifier.isStatic;
 
 import com.google.common.base.Predicate;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.reflect.TypeToken;
 
 import java.lang.annotation.Annotation;
@@ -88,4 +90,16 @@ public class TypeUtils {
         }
       };
   }
+
+  public static void checkNoInheritanceRelationships(ImmutableSet<Class<?>> resourceClasses) {
+    for (Class<?> resourceClass : resourceClasses) {
+      for (Class<?> potentialSuperclass : difference(resourceClasses, resourceClass)) {
+        checkArgument(
+            !potentialSuperclass.isAssignableFrom(resourceClass),
+            "Cannot specify resource classes with inheritance relationship: %s extends %s",
+            resourceClass,
+            potentialSuperclass);
+      }
+    }
+  }
 }

javatests/com/google/domain/registry/mapreduce/inputs/BUILD

@@ -10,13 +10,17 @@ java_library(
     srcs = glob(["*.java"]),
     deps = [
         "//java/com/google/common/base",
+        "//java/com/google/common/collect",
         "//java/com/google/domain/registry/config",
         "//java/com/google/domain/registry/mapreduce/inputs",
         "//java/com/google/domain/registry/model",
+        "//java/com/google/domain/registry/util",
         "//javatests/com/google/domain/registry/testing",
         "//third_party/java/appengine:appengine-api-testonly",
         "//third_party/java/appengine:appengine-testing",
         "//third_party/java/appengine_mapreduce2:appengine_mapreduce",
+        "//third_party/java/joda_money",
+        "//third_party/java/joda_time",
         "//third_party/java/junit",
         "//third_party/java/objectify:objectify-v4_1",
         "//third_party/java/truth",

javatests/com/google/domain/registry/mapreduce/inputs/ChildEntityInputTest.java

@@ -0,0 +1,373 @@
+// Copyright 2016 The Domain Registry Authors. All Rights Reserved.
+//
+// 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 com.google.domain.registry.mapreduce.inputs;
+
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth.assert_;
+import static com.google.domain.registry.mapreduce.inputs.EppResourceInputs.createChildEntityInput;
+import static com.google.domain.registry.model.EppResourceUtils.loadByUniqueId;
+import static com.google.domain.registry.model.index.EppResourceIndexBucket.getBucketKey;
+import static com.google.domain.registry.testing.DatastoreHelper.createTld;
+import static com.google.domain.registry.testing.DatastoreHelper.newDomainResource;
+import static com.google.domain.registry.testing.DatastoreHelper.persistActiveDomain;
+import static com.google.domain.registry.testing.DatastoreHelper.persistResource;
+import static com.google.domain.registry.testing.DatastoreHelper.persistSimpleResource;
+import static com.google.domain.registry.util.DateTimeUtils.END_OF_TIME;
+import static org.joda.money.CurrencyUnit.USD;
+
+import com.google.appengine.tools.mapreduce.InputReader;
+import com.google.common.collect.ImmutableSet;
+import com.google.domain.registry.config.TestRegistryConfig;
+import com.google.domain.registry.model.EppResource;
+import com.google.domain.registry.model.ImmutableObject;
+import com.google.domain.registry.model.billing.BillingEvent;
+import com.google.domain.registry.model.billing.BillingEvent.Reason;
+import com.google.domain.registry.model.contact.ContactResource;
+import com.google.domain.registry.model.domain.DomainResource;
+import com.google.domain.registry.model.index.EppResourceIndex;
+import com.google.domain.registry.model.reporting.HistoryEntry;
+import com.google.domain.registry.testing.AppEngineRule;
+import com.google.domain.registry.testing.ExceptionRule;
+import com.google.domain.registry.testing.RegistryConfigRule;
+
+import com.googlecode.objectify.Key;
+
+import org.joda.money.Money;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.HashSet;
+import java.util.NoSuchElementException;
+import java.util.Set;
+
+/** Tests {@link ChildEntityInput} */
+@RunWith(JUnit4.class)
+public class ChildEntityInputTest {
+
+  private static final DateTime now = DateTime.now(DateTimeZone.UTC);
+
+  @Rule
+  public final AppEngineRule appEngine = AppEngineRule.builder().withDatastore().build();
+
+  @Rule
+  public final ExceptionRule thrown = new ExceptionRule();
+
+  @Rule
+  public final RegistryConfigRule configRule = new RegistryConfigRule();
+
+  DomainResource domainA;
+  DomainResource domainB;
+  HistoryEntry domainHistoryEntryA;
+  HistoryEntry domainHistoryEntryB;
+  HistoryEntry contactHistoryEntry;
+  BillingEvent.OneTime oneTimeA;
+  BillingEvent.OneTime oneTimeB;
+  BillingEvent.Recurring recurringA;
+  BillingEvent.Recurring recurringB;
+
+  private void overrideBucketCount(final int count) {
+    configRule.override(new TestRegistryConfig() {
+      @Override
+      public int getEppResourceIndexBucketCount() {
+        return count;
+      }
+    });
+  }
+
+  private void setupResources() {
+    createTld("tld");
+    overrideBucketCount(1);
+    domainA = persistActiveDomain("a.tld");
+    domainHistoryEntryA = persistResource(
+        new HistoryEntry.Builder()
+            .setParent(domainA)
+            .setModificationTime(now)
+            .build());
+    contactHistoryEntry = persistResource(
+        new HistoryEntry.Builder()
+            .setParent(loadByUniqueId(ContactResource.class, "contact1234", now))
+            .setModificationTime(now)
+            .build());
+    oneTimeA = persistResource(
+        new BillingEvent.OneTime.Builder()
+            .setParent(domainHistoryEntryA)
+            .setReason(Reason.CREATE)
+            .setFlags(ImmutableSet.of(BillingEvent.Flag.ANCHOR_TENANT))
+            .setPeriodYears(2)
+            .setCost(Money.of(USD, 1))
+            .setEventTime(now)
+            .setBillingTime(now.plusDays(5))
+            .setClientId("TheRegistrar")
+            .setTargetId("a.tld")
+            .build());
+    recurringA = persistResource(
+        new BillingEvent.Recurring.Builder()
+            .setParent(domainHistoryEntryA)
+            .setReason(Reason.AUTO_RENEW)
+            .setEventTime(now.plusYears(1))
+            .setRecurrenceEndTime(END_OF_TIME)
+            .setClientId("TheRegistrar")
+            .setTargetId("a.tld")
+            .build());
+  }
+
+  private void setupSecondDomainResources() {
+    domainB = persistActiveDomain("b.tld");
+    domainHistoryEntryB = persistResource(
+        new HistoryEntry.Builder()
+            .setParent(domainB)
+            .setModificationTime(now)
+            .build());
+    oneTimeB = persistResource(
+        new BillingEvent.OneTime.Builder()
+            .setParent(domainHistoryEntryA)
+            .setReason(Reason.CREATE)
+            .setFlags(ImmutableSet.of(BillingEvent.Flag.ANCHOR_TENANT))
+            .setPeriodYears(2)
+            .setCost(Money.of(USD, 1))
+            .setEventTime(now)
+            .setBillingTime(now.plusDays(5))
+            .setClientId("TheRegistrar")
+            .setTargetId("a.tld")
+            .build());
+    recurringB = persistResource(
+        new BillingEvent.Recurring.Builder()
+            .setParent(domainHistoryEntryA)
+            .setReason(Reason.AUTO_RENEW)
+            .setEventTime(now.plusYears(1))
+            .setRecurrenceEndTime(END_OF_TIME)
+            .setClientId("TheRegistrar")
+            .setTargetId("a.tld")
+            .build());
+  }
+
+  @SuppressWarnings("unchecked")
+  private <T> T serializeAndDeserialize(T obj) throws Exception {
+    try (ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
+        ObjectOutputStream objectOut = new ObjectOutputStream(byteOut)) {
+      objectOut.writeObject(obj);
+      try (ByteArrayInputStream byteIn = new ByteArrayInputStream(byteOut.toByteArray());
+          ObjectInputStream objectIn = new ObjectInputStream(byteIn)) {
+        return (T) objectIn.readObject();
+      }
+    }
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_multipleParentsAndChildren() throws Exception {
+    setupResources();
+    setupSecondDomainResources();
+    Set<ImmutableObject> seen = new HashSet<>();
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(EppResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(
+            HistoryEntry.class, BillingEvent.OneTime.class, BillingEvent.Recurring.class))
+        .createReaders().get(0);
+    reader.beginShard();
+    reader.beginSlice();
+    for (int i = 0; i < 8; i++) {
+      reader.endSlice();
+      reader = serializeAndDeserialize(reader);
+      reader.beginSlice();
+      if (i == 7) {
+        thrown.expect(NoSuchElementException.class);
+      }
+      seen.add(reader.next());
+    }
+    assertThat(seen).containsExactly(
+        domainHistoryEntryA,
+        domainHistoryEntryB,
+        contactHistoryEntry,
+        oneTimeA,
+        recurringA,
+        oneTimeB,
+        recurringB);
+  }
+
+  @Test
+  public void testSuccess_childEntityInput_polymorphicBaseType() throws Exception {
+    createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(EppResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(BillingEvent.class));
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_multipleChildTypes() throws Exception {
+    setupResources();
+    Set<ImmutableObject> seen = new HashSet<>();
+
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(EppResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(
+            HistoryEntry.class, BillingEvent.OneTime.class, BillingEvent.Recurring.class))
+        .createReaders().get(0);
+
+    reader.beginShard();
+    reader.beginSlice();
+    seen.add(reader.next());
+    seen.add(reader.next());
+    seen.add(reader.next());
+    seen.add(reader.next());
+    assertThat(seen).containsExactly(
+        domainHistoryEntryA, contactHistoryEntry, oneTimeA, recurringA);
+    thrown.expect(NoSuchElementException.class);
+    reader.next();
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_filterParentTypes() throws Exception {
+    setupResources();
+    Set<ImmutableObject> seen = new HashSet<>();
+
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(ContactResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(
+            HistoryEntry.class, BillingEvent.OneTime.class, BillingEvent.Recurring.class))
+        .createReaders().get(0);
+
+    reader.beginShard();
+    reader.beginSlice();
+    seen.add(reader.next());
+    assertThat(seen).containsExactly(contactHistoryEntry);
+    thrown.expect(NoSuchElementException.class);
+    reader.next();
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_polymorphicChildFiltering() throws Exception {
+    setupResources();
+    Set<ImmutableObject> seen = new HashSet<>();
+
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(EppResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(BillingEvent.OneTime.class))
+            .createReaders().get(0);
+
+    reader.beginShard();
+    reader.beginSlice();
+    seen.add(reader.next());
+    assertThat(seen).containsExactly(oneTimeA);
+    thrown.expect(NoSuchElementException.class);
+    reader.next();
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_polymorphicChildClass() throws Exception {
+    setupResources();
+    Set<ImmutableObject> seen = new HashSet<>();
+
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(EppResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(BillingEvent.class))
+        .createReaders().get(0);
+
+    reader.beginShard();
+    reader.beginSlice();
+    seen.add(reader.next());
+    seen.add(reader.next());
+    assertThat(seen).containsExactly(oneTimeA, recurringA);
+    thrown.expect(NoSuchElementException.class);
+    reader.next();
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_noneReturned() throws Exception {
+    createTld("tld");
+    overrideBucketCount(1);
+
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(ContactResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(
+            BillingEvent.OneTime.class)).createReaders().get(0);
+
+    reader.beginShard();
+    reader.beginSlice();
+    thrown.expect(NoSuchElementException.class);
+    reader.next();
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_readerCountMatchesBucketCount() throws Exception {
+    overrideBucketCount(123);
+    assertThat(EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(DomainResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(
+            BillingEvent.OneTime.class)).createReaders()).hasSize(123);
+  }
+
+  @Test
+  public void testSuccess_childEntityReader_oneReaderPerBucket() throws Exception {
+    overrideBucketCount(3);
+    createTld("tld");
+    Set<ImmutableObject> historyEntries = new HashSet<>();
+    for (int i = 1; i <= 3; i++) {
+      DomainResource domain = persistSimpleResource(newDomainResource(i + ".tld"));
+      historyEntries.add(persistResource(
+          new HistoryEntry.Builder()
+              .setParent(domain)
+              .setModificationTime(now)
+              .setClientId(i + ".tld")
+              .build()));
+      persistResource(EppResourceIndex.create(getBucketKey(i), Key.create(domain)));
+    }
+    Set<ImmutableObject> seen = new HashSet<>();
+    for (InputReader<ImmutableObject> reader : EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(DomainResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(HistoryEntry.class)).createReaders()) {
+      reader.beginShard();
+      reader.beginSlice();
+      seen.add(reader.next());
+      try {
+        ImmutableObject o = reader.next();
+        assert_().fail("Unexpected element: " + o);
+      } catch (NoSuchElementException expected) {
+      }
+    }
+    assertThat(seen).containsExactlyElementsIn(historyEntries);
+  }
+
+
+  @Test
+  public void testSuccess_childEntityReader_survivesAcrossSerialization() throws Exception {
+    setupResources();
+    Set<ImmutableObject> seen = new HashSet<>();
+    InputReader<ImmutableObject> reader = EppResourceInputs.createChildEntityInput(
+        ImmutableSet.<Class<? extends EppResource>>of(EppResource.class),
+        ImmutableSet.<Class<? extends ImmutableObject>>of(
+            HistoryEntry.class, BillingEvent.OneTime.class, BillingEvent.Recurring.class))
+        .createReaders().get(0);
+    reader.beginShard();
+    reader.beginSlice();
+    seen.add(reader.next());
+    seen.add(reader.next());
+    reader.endSlice();
+    reader = serializeAndDeserialize(reader);
+    reader.beginSlice();
+    seen.add(reader.next());
+    seen.add(reader.next());
+    assertThat(seen).containsExactly(
+        domainHistoryEntryA, contactHistoryEntry, oneTimeA, recurringA);
+    thrown.expect(NoSuchElementException.class);
+    reader.next();
+  }
+}