Add datePredicate

dora/core/server/master/src/main/java/alluxio/master/predicate/DatePredicate.java

@@ -0,0 +1,131 @@
+package alluxio.master.predicate;
+
+import alluxio.exception.runtime.FailedPreconditionRuntimeException;
+import alluxio.master.predicate.interval.Interval;
+import alluxio.proto.journal.Job.FileFilter;
+import alluxio.wire.FileInfo;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+/**
+ * A predicate related to date of the file.
+ */
+public class DatePredicate implements FilePredicate {
+  private static final Logger LOG = LoggerFactory.getLogger(DatePredicate.class);
+  private final String mFilterName;
+  private String mValue;
+  private final Interval mInterval;
+  private final Function<FileInfo, Long> mGetter;
+
+  /**
+   * Factory for modification time predicate.
+   */
+  public static class LastModifiedDateFactory extends Factory {
+    @Override
+    public String getFilterName() {
+      return "lastModifiedDate";
+    }
+
+    @Override
+    public Function<FileInfo, Long> getTimestampGetter() {
+      return FileInfo::getLastModificationTimeMs;
+    }
+  }
+
+  /**
+   * Factory for creating instances.
+   */
+  public abstract static class Factory implements FilePredicateFactory {
+    /**
+     * @return filter name for the predicate
+     */
+    public abstract String getFilterName();
+
+    /**
+     * @return getter for the timestamp
+     */
+    public abstract Function<FileInfo, Long> getTimestampGetter();
+
+    /**
+     * Creates a {@link FilePredicate} from the string value.
+     *
+     * @param value the value from the filter
+     * @return the created predicate
+     */
+    public FilePredicate createDatePredicate(String value) {
+      return new DatePredicate(getFilterName(), value, getTimestampGetter());
+    }
+
+    @Override
+    public FilePredicate create(FileFilter filter) {
+      try {
+        if (filter.hasName() && filter.getName().equals(getFilterName())) {
+          if (filter.hasValue()) {
+            return createDatePredicate(filter.getValue());
+          }
+        }
+      } catch (Exception e) {
+        // fall through
+      }
+      return null;
+    }
+  }
+
+  /**
+   * Creates an instance.
+   *
+   * @param filterName the filter name
+   * @param value the string representation of the time span. The string should contain either
+   *              one date like "2020/03/01" to define an interval after a start date, or
+   *              two dates delimited by comma like "2000/01/01, 2020/09/01" to define an interval
+   *              between a start date and an end date
+   * @param getter a getter function that returns the timestamp to compare against
+   */
+  public DatePredicate(String filterName, String value, Function<FileInfo, Long> getter) {
+    mFilterName = filterName;
+    mValue = value;
+    mInterval = parseInterval(value);
+    mGetter = getter;
+  }
+
+  private Interval parseInterval(String stringValue) {
+    String[] values = stringValue.split(",");
+    if (values.length == 1) {
+      return Interval.after(convertDateToMs(values[0].trim()));
+    }
+    if (values.length == 2) {
+      return Interval.between(
+          convertDateToMs(values[0].trim()), convertDateToMs(values[1].trim()));
+    }
+    throw new IllegalArgumentException("Fail to parse " + stringValue);
+  }
+
+  private long convertDateToMs(String dateString) {
+    SimpleDateFormat f = new SimpleDateFormat("yyyy/MM/dd");
+    try {
+      Date d = f.parse(dateString);
+      return d.getTime();
+    } catch (Exception e) {
+      throw FailedPreconditionRuntimeException.from(e);
+    }
+  }
+
+  @Override
+  public Predicate<FileInfo> get() {
+    return FileInfo -> {
+      try {
+        Interval interval = Interval.between(0, mGetter.apply(FileInfo));
+        return mInterval.intersect(interval).isValid();
+      } catch (RuntimeException e) {
+        LOG.debug("Failed to filter: ", e);
+        return false;
+      }
+    };
+  }
+}

dora/core/server/master/src/main/resources/META-INF/services/alluxio.master.predicate.FilePredicateFactory

@@ -10,4 +10,5 @@
 #
 
 alluxio.master.predicate.TimePredicate$DateFromFileNameOlderThanFactory
-alluxio.master.predicate.TimePredicate$UnmodifiedForFactory
+alluxio.master.predicate.TimePredicate$UnmodifiedForFactory
+alluxio.master.predicate.DatePredicate$LastModifiedDateFactory

dora/core/server/master/src/test/java/alluxio/master/predicate/DatePredicateTest.java

@@ -0,0 +1,66 @@
+package alluxio.master.predicate;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import alluxio.proto.journal.Job.FileFilter;
+import alluxio.wire.FileInfo;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+/**
+ * Unit tests for {@link DatePredicate}.
+ */
+public class DatePredicateTest {
+  /**
+   * The exception expected to be thrown.
+   */
+  @Rule
+  public final ExpectedException mThrown = ExpectedException.none();
+
+  @Test
+  public void testDatePredicateFactories() {
+    FileFilter filter = FileFilter.newBuilder().setName("lastModifiedDate")
+        .setValue("2020/01/01").build();
+    long timestamp = System.currentTimeMillis();
+    FileInfo info = new FileInfo();
+    info.setLastModificationTimeMs(timestamp);
+
+    assertTrue(FilePredicate.create(filter).get().test(info));
+
+    //2019/09/05
+    info.setLastModificationTimeMs(1568523242000L);
+    assertFalse(FilePredicate.create(filter).get().test(info));
+  }
+
+  @Test
+  public void testDatePredicateInterval() {
+    FileFilter filter = FileFilter.newBuilder().setName("lastModifiedDate")
+        .setValue("2020/01/01, 2023/09/14").build();
+    FileInfo info = new FileInfo();
+    //2021/09/15
+    info.setLastModificationTimeMs(1631681642000L);
+
+    assertTrue(FilePredicate.create(filter).get().test(info));
+
+    //2019/09/05
+    info.setLastModificationTimeMs(1568523242000L);
+    assertFalse(FilePredicate.create(filter).get().test(info));
+  }
+
+  @Test
+  public void testInvalidDate() {
+    FileFilter filter = FileFilter.newBuilder().setName("lastModifiedDate")
+        .setValue("2020-01-01, 2023-09-14").build();
+    FileInfo info = new FileInfo();
+    //2021/09/15
+    info.setLastModificationTimeMs(1631681642000L);
+
+    mThrown.expect(UnsupportedOperationException.class);
+    mThrown.expectMessage("Invalid filter name: ");
+
+    FilePredicate.create(filter).get().test(info);
+  }
+}