apache · rdblue · Jan 22, 2021 · Nov 2, 2020 · Dec 29, 2020 · Dec 29, 2020
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Dates.java b/api/src/main/java/org/apache/iceberg/transforms/Dates.java
@@ -20,7 +20,7 @@
 package org.apache.iceberg.transforms;
 
 import java.time.Instant;
-import java.time.OffsetDateTime;
+import java.time.LocalDate;
 import java.time.ZoneOffset;
 import java.time.temporal.ChronoUnit;
 import org.apache.iceberg.expressions.BoundPredicate;
@@ -36,7 +36,7 @@ enum Dates implements Transform<Integer, Integer> {
   MONTH(ChronoUnit.MONTHS, "month"),
   DAY(ChronoUnit.DAYS, "day");
 
-  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
+  private static final LocalDate EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC).toLocalDate();
   private final ChronoUnit granularity;
   private final String name;
 
@@ -55,7 +55,15 @@ public Integer apply(Integer days) {
       return days;
     }
 
-    return (int) granularity.between(EPOCH, EPOCH.plusDays(days));
+    if (days >= 0) {
+      LocalDate date = EPOCH.plusDays(days);
+      return (int) granularity.between(EPOCH, date);
+    } else {
+      // add 1 day to the value to account for the case where there is exactly 1 unit between the date and epoch
+      // because the result will always be decremented.
+      LocalDate date = EPOCH.plusDays(days + 1);
+      return (int) granularity.between(EPOCH, date) - 1;
+    }
   }
 
   @Override
@@ -99,11 +107,24 @@ public UnboundPredicate<Integer> project(String fieldName, BoundPredicate<Intege
 
     if (pred.isUnaryPredicate()) {
       return Expressions.predicate(pred.op(), fieldName);
+
     } else if (pred.isLiteralPredicate()) {
-      return ProjectionUtil.truncateInteger(fieldName, pred.asLiteralPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.truncateInteger(fieldName, pred.asLiteralPredicate(), this);
+      if (this != DAY) {
+        return ProjectionUtil.fixInclusiveTimeProjection(projected);
+      }
+
+      return projected;
+
     } else if (pred.isSetPredicate() && pred.op() == Expression.Operation.IN) {
-      return ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      if (this != DAY) {
+        return ProjectionUtil.fixInclusiveTimeProjection(projected);
+      }
+
+      return projected;
     }
+
     return null;
   }
 
@@ -115,11 +136,25 @@ public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<
 
     if (pred.isUnaryPredicate()) {
       return Expressions.predicate(pred.op(), fieldName);
+
     } else if (pred.isLiteralPredicate()) {
-      return ProjectionUtil.truncateIntegerStrict(fieldName, pred.asLiteralPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.truncateIntegerStrict(
+          fieldName, pred.asLiteralPredicate(), this);
+      if (this != DAY) {
+        return ProjectionUtil.fixStrictTimeProjection(projected);
+      }
+
+      return projected;
+
     } else if (pred.isSetPredicate() && pred.op() == Expression.Operation.NOT_IN) {
-      return ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      if (this != DAY) {
+        return ProjectionUtil.fixStrictTimeProjection(projected);
+      }
+
+      return projected;
     }
+
     return null;
   }
 

diff --git a/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java b/api/src/main/java/org/apache/iceberg/transforms/ProjectionUtil.java
@@ -21,14 +21,17 @@
 
 import java.math.BigDecimal;
 import java.math.BigInteger;
+import java.util.Set;
 import org.apache.iceberg.expressions.BoundLiteralPredicate;
 import org.apache.iceberg.expressions.BoundPredicate;
 import org.apache.iceberg.expressions.BoundSetPredicate;
 import org.apache.iceberg.expressions.BoundTransform;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.expressions.UnboundPredicate;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 
 import static org.apache.iceberg.expressions.Expressions.predicate;
 
@@ -254,4 +257,145 @@ static <S, T> UnboundPredicate<T> transformSet(String fieldName,
     return predicate(predicate.op(), fieldName,
         Iterables.transform(predicate.asSetPredicate().literalSet(), transform::apply));
   }
+
+  /**
+   * Fixes an inclusive projection to account for incorrectly transformed values.
+   * <p>
+   * A bug in 0.10.0 and earlier caused negative values to be incorrectly transformed by date and timestamp transforms
+   * to 1 larger than the correct value. For example, day(1969-12-31 10:00:00) produced 0 instead of -1. To read data
+   * written by versions with this bug, this method adjusts the inclusive projection. The current inclusive projection
+   * is correct, so this modifies the "correct" projection when needed. For example, < day(1969-12-31 10:00:00) will
+   * produce <= -1 (= 1969-12-31) and is adjusted to <= 0 (= 1969-01-01) because the incorrect transformed value was 0.
+   */
+  static UnboundPredicate<Integer> fixInclusiveTimeProjection(UnboundPredicate<Integer> projected) {
+    if (projected == null) {
+      return projected;
+    }
+
+    // adjust the predicate for values that were 1 larger than the correct transformed value
+    switch (projected.op()) {
+      case LT:
+        if (projected.literal().value() < 0) {
+          return Expressions.lessThan(projected.term(), projected.literal().value() + 1);
+        }
+
+        return projected;
+
+      case LT_EQ:
+        if (projected.literal().value() < 0) {
+          return Expressions.lessThanOrEqual(projected.term(), projected.literal().value() + 1);
+        }
+
+        return projected;
+
+      case GT:
+      case GT_EQ:
+        // incorrect projected values are already greater than the bound for GT, GT_EQ
+        return projected;
+
+      case EQ:
+        if (projected.literal().value() < 0) {
+          // match either the incorrect value (projectedValue + 1) or the correct value (projectedValue)
+          return Expressions.in(projected.term(), projected.literal().value(), projected.literal().value() + 1);
+        }
+
+        return projected;
+
+      case IN:
+        Set<Integer> fixedSet = Sets.newHashSet();
+        boolean hasNegativeValue = false;
+        for (Literal<Integer> lit : projected.literals()) {
+          Integer value = lit.value();
+          fixedSet.add(value);
+          if (value < 0) {
+            hasNegativeValue = true;
+            fixedSet.add(value + 1);
+          }
+        }
+
+        if (hasNegativeValue) {
+          return Expressions.in(projected.term(), fixedSet);
+        }
+
+        return projected;
+
+      case NOT_IN:
+      case NOT_EQ:
+        // there is no inclusive projection for NOT_EQ and NOT_IN
+        return null;
+
+      default:
+        return projected;
+    }
+  }
+
+  /**
+   * Fixes a strict projection to account for incorrectly transformed values.
+   * <p>
+   * A bug in 0.10.0 and earlier caused negative values to be incorrectly transformed by date and timestamp transforms
+   * to 1 larger than the correct value. For example, day(1969-12-31 10:00:00) produced 0 instead of -1. To read data
+   * written by versions with this bug, this method adjusts the strict projection.
+   */
+  static UnboundPredicate<Integer> fixStrictTimeProjection(UnboundPredicate<Integer> projected) {
+    if (projected == null) {
+      return null;
+    }
+
+    switch (projected.op()) {
+      case LT:
+      case LT_EQ:
+        // the correct bound is a correct strict projection for the incorrectly transformed values.
+        return projected;
+
+      case GT:
+        // GT and GT_EQ need to be adjusted because values that do not match the predicate may have been transformed
+        // into partition values that match the projected predicate. For example, >= month(1969-11-31) is > -2, but
+        // 1969-10-31 was previously transformed to month -2 instead of -3. This must use the more strict value.
+        if (projected.literal().value() <= 0) {
+          return Expressions.greaterThan(projected.term(), projected.literal().value() + 1);
+        }
+
+        return projected;
+
+      case GT_EQ:
+        if (projected.literal().value() <= 0) {
+          return Expressions.greaterThanOrEqual(projected.term(), projected.literal().value() + 1);
+        }
+
+        return projected;
+
+      case EQ:
+      case IN:
+        // there is no strict projection for EQ and IN
+        return null;
+
+      case NOT_EQ:
+        if (projected.literal().value() < 0) {
+          return Expressions.notIn(projected.term(), projected.literal().value(), projected.literal().value() + 1);
+        }
+
+        return projected;
+
+      case NOT_IN:
+        Set<Integer> fixedSet = Sets.newHashSet();
+        boolean hasNegativeValue = false;
+        for (Literal<Integer> lit : projected.literals()) {
+          Integer value = lit.value();
+          fixedSet.add(value);
+          if (value < 0) {
+            hasNegativeValue = true;
+            fixedSet.add(value + 1);
+          }
+        }
+
+        if (hasNegativeValue) {
+          return Expressions.notIn(projected.term(), fixedSet);
+        }
+
+        return projected;
+
+      default:
+        return null;
+    }
+  }
 }
diff --git a/api/src/main/java/org/apache/iceberg/transforms/Timestamps.java b/api/src/main/java/org/apache/iceberg/transforms/Timestamps.java
@@ -52,12 +52,23 @@ public Integer apply(Long timestampMicros) {
       return null;
     }
 
-    // discards fractional seconds, not needed for calculation
-    OffsetDateTime timestamp = Instant
-        .ofEpochSecond(timestampMicros / 1_000_000)
-        .atOffset(ZoneOffset.UTC);
-
-    return (int) granularity.between(EPOCH, timestamp);
+    if (timestampMicros >= 0) {
+      OffsetDateTime timestamp = Instant
+          .ofEpochSecond(
+              Math.floorDiv(timestampMicros, 1_000_000),
+              Math.floorMod(timestampMicros, 1_000_000) * 1000)
+          .atOffset(ZoneOffset.UTC);
+      return (int) granularity.between(EPOCH, timestamp);
+    } else {
+      // add 1 micro to the value to account for the case where there is exactly 1 unit between the timestamp and epoch
+      // because the result will always be decremented.
+      OffsetDateTime timestamp = Instant
+          .ofEpochSecond(
+              Math.floorDiv(timestampMicros, 1_000_000),
+              Math.floorMod(timestampMicros + 1, 1_000_000) * 1000)
+          .atOffset(ZoneOffset.UTC);
+      return (int) granularity.between(EPOCH, timestamp) - 1;
+    }
   }
 
   @Override
@@ -101,11 +112,16 @@ public UnboundPredicate<Integer> project(String fieldName, BoundPredicate<Long>
 
     if (pred.isUnaryPredicate()) {
       return Expressions.predicate(pred.op(), fieldName);
+
     } else if (pred.isLiteralPredicate()) {
-      return ProjectionUtil.truncateLong(fieldName, pred.asLiteralPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.truncateLong(fieldName, pred.asLiteralPredicate(), this);
+      return ProjectionUtil.fixInclusiveTimeProjection(projected);
+
     } else if (pred.isSetPredicate() && pred.op() == Expression.Operation.IN) {
-      return ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      return ProjectionUtil.fixInclusiveTimeProjection(projected);
     }
+
     return null;
   }
 
@@ -117,11 +133,17 @@ public UnboundPredicate<Integer> projectStrict(String fieldName, BoundPredicate<
 
     if (pred.isUnaryPredicate()) {
       return Expressions.predicate(pred.op(), fieldName);
+
     } else if (pred.isLiteralPredicate()) {
-      return ProjectionUtil.truncateLongStrict(fieldName, pred.asLiteralPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.truncateLongStrict(
+          fieldName, pred.asLiteralPredicate(), this);
+      return ProjectionUtil.fixStrictTimeProjection(projected);
+
     } else if (pred.isSetPredicate() && pred.op() == Expression.Operation.NOT_IN) {
-      return ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      UnboundPredicate<Integer> projected = ProjectionUtil.transformSet(fieldName, pred.asSetPredicate(), this);
+      return ProjectionUtil.fixStrictTimeProjection(projected);
     }
+
     return null;
   }
 

diff --git a/api/src/main/java/org/apache/iceberg/transforms/TransformUtil.java b/api/src/main/java/org/apache/iceberg/transforms/TransformUtil.java
@@ -41,7 +41,8 @@ static String humanYear(int yearOrdinal) {
   }
 
   static String humanMonth(int monthOrdinal) {
-    return String.format("%04d-%02d", EPOCH_YEAR + (monthOrdinal / 12), 1 + (monthOrdinal % 12));
+    return String.format("%04d-%02d",
+        EPOCH_YEAR + Math.floorDiv(monthOrdinal, 12), 1 + Math.floorMod(monthOrdinal, 12));
   }
 
   static String humanDay(int dayOrdinal) {

diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestStringLiteralConversions.java b/api/src/test/java/org/apache/iceberg/expressions/TestStringLiteralConversions.java
@@ -57,6 +57,22 @@ public void testStringToDateLiteral() {
     Assert.assertEquals("Date should match", avroValue, (int) date.value());
   }
 
+  @Test
+  public void testNegativeStringToDateLiteral() {
+    Literal<CharSequence> dateStr = Literal.of("1969-12-30");
+    Literal<Integer> date = dateStr.to(Types.DateType.get());
+
+    // use Avro's date conversion to validate the result
+    Schema avroSchema = LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT));
+    TimeConversions.DateConversion avroConversion = new TimeConversions.DateConversion();
+    int avroValue = avroConversion.toInt(
+        LocalDate.of(1969, 12, 30),
+        avroSchema, avroSchema.getLogicalType());
+
+    Assert.assertEquals("Date should be -2", -2, (int) date.value());
+    Assert.assertEquals("Date should match", avroValue, (int) date.value());
+  }
+
   @Test
   public void testStringToTimeLiteral() {
     // use Avro's time conversion to validate the result
@@ -106,6 +122,43 @@ public void testStringToTimestampLiteral() {
         avroValue, (long) timestamp.value());
   }
 
+  @Test
+  public void testNegativeStringToTimestampLiteral() {
+    // use Avro's timestamp conversion to validate the result
+    Schema avroSchema = LogicalTypes.timestampMicros().addToSchema(Schema.create(Schema.Type.LONG));
+    TimeConversions.TimestampMicrosConversion avroConversion =
+        new TimeConversions.TimestampMicrosConversion();
+
+    // Timestamp with explicit UTC offset, +00:00
+    Literal<CharSequence> timestampStr = Literal.of("1969-12-31T23:59:58.999999+00:00");
+    Literal<Long> timestamp = timestampStr.to(Types.TimestampType.withZone());
+    long avroValue = avroConversion.toLong(
+        LocalDateTime.of(1969, 12, 31, 23, 59, 58, 999999 * 1_000).toInstant(ZoneOffset.UTC),
+        avroSchema, avroSchema.getLogicalType());
+
+    Assert.assertEquals("Timestamp should match", avroValue, (long) timestamp.value());
+    Assert.assertEquals("Timestamp should be -1_000_001", -1_000_001, (long) timestamp.value());
+
+    // Timestamp without an explicit zone should be UTC (equal to the previous converted value)
+    timestampStr = Literal.of("1969-12-31T23:59:58.999999");
+    timestamp = timestampStr.to(Types.TimestampType.withoutZone());
+
+    Assert.assertEquals("Timestamp without zone should match UTC",
+        avroValue, (long) timestamp.value());
+
+    // Timestamp with an explicit offset should be adjusted to UTC
+    timestampStr = Literal.of("1969-12-31T16:59:58.999999-07:00");
+    timestamp = timestampStr.to(Types.TimestampType.withZone());
+    avroValue = avroConversion.toLong(
+        LocalDateTime.of(1969, 12, 31, 23, 59, 58, 999999 * 1_000).toInstant(ZoneOffset.UTC),
+        avroSchema, avroSchema.getLogicalType());
+
+    Assert.assertEquals("Timestamp without zone should match UTC",
+        avroValue, (long) timestamp.value());
+    Assert.assertEquals("Timestamp without zone should be -1_000_001", -1_000_001, (long) timestamp.value());
+
+  }
+
   @Test(expected = DateTimeException.class)
   public void testTimestampWithZoneWithoutZoneInLiteral() {
     // Zone must be present in literals when converting to timestamp with zone