Strict projection for dates timestamps

api/src/main/java/org/apache/iceberg/transforms/Dates.java

@@ -73,8 +73,11 @@ public UnboundPredicate<Integer> project(String fieldName, BoundPredicate<Intege
   }
 
   @Override
-  public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<Integer> predicate) {
-    return null;
+  public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<Integer> pred) {
+    if (pred.op() == NOT_NULL || pred.op() == IS_NULL) {
+      return Expressions.predicate(pred.op(), fieldName);
+    }
+    return ProjectionUtil.truncateIntegerStrictToInteger(fieldName, pred, this);
   }
 
   @Override

api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java

@@ -52,6 +52,84 @@ static <T> UnboundPredicate<T> truncateInteger(
     }
   }
 
+  static UnboundPredicate<Integer> truncateIntegerStrictToInteger(
+      String name, BoundPredicate<Integer> pred, Transform<Integer, Integer> transform) {
+    Integer boundary = pred.literal().value();
+    switch (pred.op()) {
+      case LT:
+        // Checking if the date is at the lower boundary
+        if (!transform.apply(boundary - 1).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.LT_EQ, name, transform.apply(boundary) - 1);
+        } else {
+          return null;
+        }
+      case LT_EQ:
+        // Checking if the date is at the upper boundary
+        if (!transform.apply(boundary + 1).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.LT_EQ, name, transform.apply(boundary));
+        } else {
+          return null;
+        }
+      case GT:
+        // Checking if the date is at the upper boundary
+        if (!transform.apply(boundary + 1).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.GT_EQ, name, transform.apply(boundary) + 1);
+        } else {
+          return null;
+        }
+      case GT_EQ:
+        // Checking if the date is at the lower boundary
+        if (!transform.apply(boundary - 1).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.GT_EQ, name, transform.apply(boundary));
+        } else {
+          return null;
+        }
+      case NOT_EQ:
+        return predicate(Expression.Operation.NOT_EQ, name, transform.apply(boundary));
+      default:
+        return null;
+    }
+  }
+
+  static UnboundPredicate<Integer> truncateLongStrictToInteger(
+      String name, BoundPredicate<Long> pred, Transform<Long, Integer> transform) {
+    Long boundary = pred.literal().value();
+    switch (pred.op()) {
+      case LT:
+        // Checking if the timestamp is at the lower boundary
+        if (!transform.apply(boundary - 1L).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.LT_EQ, name, transform.apply(boundary) - 1);
+        } else {
+          return null;
+        }
+      case LT_EQ:
+        // Checking if the timestamp is at the upper boundary
+        if (!transform.apply(boundary + 1L).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.LT_EQ, name, transform.apply(boundary));
+        } else {
+          return null;
+        }
+      case GT:
+        // Checking if the timestamp is at the upper boundary
+        if (!transform.apply(boundary + 1L).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.GT_EQ, name, transform.apply(boundary) + 1);
+        } else {
+          return null;
+        }
+      case GT_EQ:
+        // Checking if the timestamp is at the lower boundary
+        if (!transform.apply(boundary - 1L).equals(transform.apply(boundary))) {
+          return predicate(Expression.Operation.GT_EQ, name, transform.apply(boundary));
+        } else {
+          return null;
+        }
+      case NOT_EQ:
+        return predicate(Expression.Operation.NOT_EQ, name, transform.apply(boundary));
+      default:
+        return null;
+    }
+  }
+
   static <T> UnboundPredicate<T> truncateLong(
       String name, BoundPredicate<Long> pred, Transform<Long, T> transform) {
     long boundary = pred.literal().value();

api/src/main/java/org/apache/iceberg/transforms/Timestamps.java

@@ -76,8 +76,11 @@ public UnboundPredicate<Integer> project(String fieldName, BoundPredicate<Long>
   }
 
   @Override
-  public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<Long> predicate) {
-    return null;
+  public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<Long> pred) {
+    if (pred.op() == NOT_NULL || pred.op() == IS_NULL) {
+      return Expressions.predicate(pred.op(), fieldName);
+    }
+    return ProjectionUtil.truncateLongStrictToInteger(fieldName, pred, this);
   }
 
   @Override

api/src/test/java/org/apache/iceberg/transforms/TestDatesProjection.java

@@ -0,0 +1,202 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.transforms;
+
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.expressions.UnboundPredicate;
+import org.apache.iceberg.types.Types;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.apache.iceberg.TestHelpers.assertAndUnwrapUnbound;
+import static org.apache.iceberg.expressions.Expressions.equal;
+import static org.apache.iceberg.expressions.Expressions.greaterThan;
+import static org.apache.iceberg.expressions.Expressions.greaterThanOrEqual;
+import static org.apache.iceberg.expressions.Expressions.lessThan;
+import static org.apache.iceberg.expressions.Expressions.lessThanOrEqual;
+import static org.apache.iceberg.expressions.Expressions.notEqual;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+
+public class TestDatesProjection {
+  private static final Types.DateType TYPE = Types.DateType.get();
+  private static final Schema SCHEMA = new Schema(optional(1, "date", TYPE));
+
+  public void assertProjectionStrict(PartitionSpec spec, UnboundPredicate<?> filter,
+                                     Expression.Operation expectedOp, String expectedLiteral) {
+
+    Expression projection = Projections.strict(spec).project(filter);
+    UnboundPredicate<?> predicate = assertAndUnwrapUnbound(projection);
+
+    Assert.assertEquals(predicate.op(), expectedOp);
+
+    Literal literal = predicate.literal();
+    Dates transform = (Dates) spec.getFieldsBySourceId(1).get(0).transform();
+    String output = transform.toHumanString((int) literal.value());
+    Assert.assertEquals(expectedLiteral, output);
+
+  }
+
+  public void assertProjectionInclusive(PartitionSpec spec, UnboundPredicate<?> filter,
+                                        Expression.Operation expectedOp, String expectedLiteral) {
+    Expression projection = Projections.inclusive(spec).project(filter);
+    UnboundPredicate<?> predicate = assertAndUnwrapUnbound(projection);
+
+    Assert.assertEquals(predicate.op(), expectedOp);
+
+    Literal literal = predicate.literal();
+    Dates transform = (Dates) spec.getFieldsBySourceId(1).get(0).transform();
+    String output = transform.toHumanString((int) literal.value());
+    Assert.assertEquals(expectedLiteral, output);
+
+  }
+
+  @Test
+  public void testMonthStrictLowerBound() {
+    Integer date = (Integer) Literal.of("2017-01-01").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).month("date").build();
+
+    assertProjectionStrict(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2016-12");
+    // bound should include the same month for lower bound
+    assertProjectionStrict(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017-01");
+    assertProjectionStrict(spec, notEqual("date", date), Expression.Operation.NOT_EQ, "2017-01");
+
+  }
+
+  @Test
+  public void testMonthStrictUpperBound() {
+    Integer date = (Integer) Literal.of("2017-12-31").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).month("date").build();
+
+    // bound should include the same month for upper bound
+    assertProjectionStrict(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017-12");
+    assertProjectionStrict(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2018-01");
+    assertProjectionStrict(spec, notEqual("date", date), Expression.Operation.NOT_EQ, "2017-12");
+
+  }
+
+  @Test
+  public void testMonthInclusiveLowerBound() {
+    Integer date = (Integer) Literal.of("2017-12-01").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).month("date").build();
+
+    assertProjectionInclusive(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2017-11");
+    assertProjectionInclusive(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017-12");
+    assertProjectionInclusive(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2017-12");
+    assertProjectionInclusive(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017-12");
+    assertProjectionInclusive(spec, equal("date", date), Expression.Operation.EQ, "2017-12");
+
+  }
+
+  @Test
+  public void testMonthInclusiveUpperBound() {
+    Integer date = (Integer) Literal.of("2017-12-31").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).month("date").build();
+
+    assertProjectionInclusive(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2017-12");
+    assertProjectionInclusive(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017-12");
+    assertProjectionInclusive(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2018-01");
+    assertProjectionInclusive(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017-12");
+    assertProjectionInclusive(spec, equal("date", date), Expression.Operation.EQ, "2017-12");
+
+  }
+
+  @Test
+  public void testDayStrict() {
+    Integer date = (Integer) Literal.of("2017-01-01").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).day("date").build();
+
+    assertProjectionStrict(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2016-12-31");
+    // should be the same date for <=
+    assertProjectionStrict(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017-01-01");
+    assertProjectionStrict(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2017-01-02");
+    // should be the same date for >=
+    assertProjectionStrict(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017-01-01");
+    assertProjectionStrict(spec, notEqual("date", date), Expression.Operation.NOT_EQ, "2017-01-01");
+
+  }
+
+  @Test
+  public void testDayInclusive() {
+    Integer date = (Integer) Literal.of("2017-12-01").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).day("date").build();
+
+    assertProjectionInclusive(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2017-11-30");
+    assertProjectionInclusive(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017-12-01");
+    assertProjectionInclusive(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2017-12-02");
+    assertProjectionInclusive(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017-12-01");
+    assertProjectionInclusive(spec, equal("date", date), Expression.Operation.EQ, "2017-12-01");
+
+  }
+
+  @Test
+  public void testYearStrictLowerBound() {
+    Integer date = (Integer) Literal.of("2017-01-01").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).year("date").build();
+
+    assertProjectionStrict(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2016");
+    // bound should include the same year for lower bound
+    assertProjectionStrict(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017");
+    assertProjectionStrict(spec, notEqual("date", date), Expression.Operation.NOT_EQ, "2017");
+
+  }
+
+  @Test
+  public void testYearStrictUpperBound() {
+    Integer date = (Integer) Literal.of("2017-12-31").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).year("date").build();
+
+    // bound should include the same year for upper bound
+    assertProjectionStrict(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017");
+    assertProjectionStrict(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2018");
+    assertProjectionStrict(spec, notEqual("date", date), Expression.Operation.NOT_EQ, "2017");
+
+  }
+
+  @Test
+  public void testYearInclusiveLowerBound() {
+    Integer date = (Integer) Literal.of("2017-01-01").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).year("date").build();
+
+    assertProjectionInclusive(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2016");
+    assertProjectionInclusive(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017");
+    assertProjectionInclusive(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2017");
+    assertProjectionInclusive(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017");
+    assertProjectionInclusive(spec, equal("date", date), Expression.Operation.EQ, "2017");
+
+  }
+
+  @Test
+  public void testYearInclusiveUpperBound() {
+    Integer date = (Integer) Literal.of("2017-12-31").to(TYPE).value();
+    PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).year("date").build();
+
+    assertProjectionInclusive(spec, lessThan("date", date), Expression.Operation.LT_EQ, "2017");
+    assertProjectionInclusive(spec, lessThanOrEqual("date", date), Expression.Operation.LT_EQ, "2017");
+    assertProjectionInclusive(spec, greaterThan("date", date), Expression.Operation.GT_EQ, "2018");
+    assertProjectionInclusive(spec, greaterThanOrEqual("date", date), Expression.Operation.GT_EQ, "2017");
+    assertProjectionInclusive(spec, equal("date", date), Expression.Operation.EQ, "2017");
+
+  }
+
+}