Support hive non string partition cols

api/src/main/java/org/apache/iceberg/expressions/And.java

@@ -19,7 +19,10 @@
 
 package org.apache.iceberg.expressions;
 
-public class And implements Expression {
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class And implements Expression, Bound<Boolean> {
   private final Expression left;
   private final Expression right;
 
@@ -47,6 +50,21 @@ public Expression negate() {
     return Expressions.or(left.negate(), right.negate());
   }
 
+  @Override
+  public BoundReference<?> ref() {
+    return null;
+  }
+
+  @Override
+  public Boolean eval(StructLike struct) {
+    Preconditions.checkNotNull(left, "Left expression cannot be null.");
+    Preconditions.checkNotNull(right, "Right expression cannot be null.");
+    if (!(left instanceof Bound) || !(right instanceof Bound)) {
+      throw new IllegalStateException("Bound predicate not expected");
+    }
+    return ((Bound<Boolean>) left).eval(struct) && ((Bound<Boolean>) right).eval(struct);
+  }
+
   @Override
   public String toString() {
     return String.format("(%s and %s)", left, right);

api/src/main/java/org/apache/iceberg/expressions/Or.java

@@ -19,7 +19,10 @@
 
 package org.apache.iceberg.expressions;
 
-public class Or implements Expression {
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class Or implements Expression, Bound<Boolean> {
   private final Expression left;
   private final Expression right;
 
@@ -47,6 +50,21 @@ public Expression negate() {
     return Expressions.and(left.negate(), right.negate());
   }
 
+  @Override
+  public BoundReference<?> ref() {
+    return null;
+  }
+
+  @Override
+  public Boolean eval(StructLike struct) {
+    Preconditions.checkNotNull(left, "Left expression cannot be null.");
+    Preconditions.checkNotNull(right, "Right expression cannot be null.");
+    if (!(left instanceof Bound) || !(right instanceof Bound)) {
+      throw new IllegalStateException("Bound predicate not expected");
+    }
+    return ((Bound<Boolean>) left).eval(struct) || ((Bound<Boolean>) right).eval(struct);
+  }
+
   @Override
   public String toString() {
     return String.format("(%s or %s)", left, right);

api/src/main/java/org/apache/iceberg/types/Conversions.java

@@ -71,6 +71,8 @@ public static Object fromPartitionString(Type type, String asString) {
         return new BigDecimal(asString);
       case DATE:
         return Literal.of(asString).to(Types.DateType.get()).value();
+      case TIMESTAMP:
+        return Literal.of(asString).to(Types.TimestampType.withoutZone()).value();
       default:
         throw new UnsupportedOperationException(
             "Unsupported type for fromPartitionString: " + type);

hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/HiveExpressions.java

@@ -19,6 +19,10 @@
 
 package org.apache.iceberg.hive.legacy;
 
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -29,6 +33,7 @@
 import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.expressions.UnboundPredicate;
 import org.apache.iceberg.expressions.UnboundTerm;
+import org.apache.iceberg.types.Type;
 
 
 class HiveExpressions {
@@ -238,6 +243,7 @@ public <T> Expression predicate(UnboundPredicate<T> pred) {
   }
 
   private static class ExpressionToPartitionFilterString extends ExpressionVisitors.ExpressionVisitor<String> {
+    private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
     private static final ExpressionToPartitionFilterString INSTANCE = new ExpressionToPartitionFilterString();
 
     private ExpressionToPartitionFilterString() {
@@ -274,26 +280,29 @@ public String or(String leftResult, String rightResult) {
 
     @Override
     public <T> String predicate(BoundPredicate<T> pred) {
-      throw new IllegalStateException("Bound predicate not expected: " + pred.getClass().getName());
-    }
-
-    @Override
-    public <T> String predicate(UnboundPredicate<T> pred) {
       switch (pred.op()) {
         case LT:
         case LT_EQ:
         case GT:
         case GT_EQ:
         case EQ:
         case NOT_EQ:
-          return getBinaryExpressionString(pred.ref().name(), pred.op(), pred.literal());
+          return getBinaryExpressionString(pred);
         default:
           throw new IllegalStateException("Unexpected operator in Hive partition filter string: " + pred.op());
       }
     }
 
-    private <T> String getBinaryExpressionString(String columnName, Expression.Operation op, Literal<T> lit) {
-      return String.format("( %s %s %s )", columnName, getOperationString(op), getLiteralValue(lit));
+    @Override
+    public <T> String predicate(UnboundPredicate<T> pred) {
+      throw new IllegalStateException("Unbound predicate not expected: " + pred.getClass().getName());
+    }
+
+    private <T> String getBinaryExpressionString(BoundPredicate<T> pred) {
+      String columnName = pred.ref().field().name();
+      String opName = getOperationString(pred.op());
+      String litValue = getLiteralValue(pred.asLiteralPredicate().literal(), pred.ref().type());
+      return String.format("( %s %s %s )", columnName, opName, litValue);
     }
 
     private String getOperationString(Expression.Operation op) {
@@ -315,8 +324,11 @@ private String getOperationString(Expression.Operation op) {
       }
     }
 
-    private <T> String getLiteralValue(Literal<T> lit) {
+    private <T> String getLiteralValue(Literal<T> lit, Type type) {
       Object value = lit.value();
+      if (type.typeId() == Type.TypeID.DATE) {
+        value = EPOCH.plus((Integer) value, ChronoUnit.DAYS).toLocalDate().toString();
+      }
       if (value instanceof String) {
         String escapedString = ((String) value).replace("'", "\\'");
         return String.format("'%s'", escapedString);

hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/LegacyHiveTableOperations.java

@@ -19,12 +19,19 @@
 
 package org.apache.iceberg.hive.legacy;
 
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.iceberg.BaseMetastoreTableOperations;
 import org.apache.iceberg.DataFile;
@@ -36,6 +43,9 @@
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.Bound;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.hadoop.HadoopFileIO;
@@ -48,6 +58,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -109,7 +120,7 @@ protected void doRefresh() {
   /**
    * Returns an {@link Iterable} of {@link Iterable}s of {@link DataFile}s which belong to the current table and
    * match the partition predicates from the given expression.
-   *
+   * <p>
    * Each element in the outer {@link Iterable} maps to an {@link Iterable} of {@link DataFile}s originating from the
    * same directory
    */
@@ -170,9 +181,11 @@ private List<DirectoryInfo> getDirectoryInfosByFilter(Expression expression) {
           .map(id -> current().schema().findColumnName(id))
           .collect(Collectors.toSet());
       Expression simplified = HiveExpressions.simplifyPartitionFilter(expression, partitionColumnNames);
+      Types.StructType partitionSchema = current().spec().partitionType();
       LOG.info("Simplified expression for {}.{} to {}", databaseName, tableName, simplified);
 
-      final List<Partition> partitions;
+      List<Partition> partitions;
+      Expression boundExpression;
       if (simplified.equals(Expressions.alwaysFalse())) {
         // If simplifyPartitionFilter returns FALSE, no partitions are going to match the filter expression
         partitions = ImmutableList.of();
@@ -181,10 +194,47 @@ private List<DirectoryInfo> getDirectoryInfosByFilter(Expression expression) {
         partitions = metaClients.run(client -> client.listPartitionsByFilter(
             databaseName, tableName, null, (short) -1));
       } else {
-        String partitionFilterString = HiveExpressions.toPartitionFilterString(simplified);
+        boundExpression = Binder.bind(partitionSchema, simplified, false);
+        String partitionFilterString = HiveExpressions.toPartitionFilterString(boundExpression);
         LOG.info("Listing partitions for {}.{} with filter string: {}", databaseName, tableName, partitionFilterString);
-        partitions = metaClients.run(
-            client -> client.listPartitionsByFilter(databaseName, tableName, partitionFilterString, (short) -1));
+        try {
+          // We first try to use HMS API call to get the filtered partitions.
+          partitions = metaClients.run(
+              client -> client.listPartitionsByFilter(databaseName, tableName, partitionFilterString, (short) -1));
+        } catch (MetaException e) {
+          // If the above HMS call fails, we here try to do the partition filtering ourselves,
+          // by evaluating all the partitions we got back from HMS against the boundExpression,
+          // if the evaluation results in true, we include such partition, if false, we filter.
+          List<Partition> allPartitions = metaClients.run(
+              client -> client.listPartitionsByFilter(databaseName, tableName, null, (short) -1));
+          partitions = allPartitions.stream().filter(partition -> {
+            GenericRecord record = GenericRecord.create(partitionSchema);
+            for (int i = 0; i < record.size(); i++) {
+              String value = partition.getValues().get(i);
+              switch (partitionSchema.fields().get(i).type().typeId()) {
+                case DATE:
+                  record.set(i,
+                      (int) LocalDate.parse(value).toEpochDay());
+                  break;
+                case TIMESTAMP:
+                  // This format seems to be matching the hive timestamp column partition string literal value
+                  record.set(i,
+                      LocalDateTime.parse(value,
+                          new DateTimeFormatterBuilder()
+                              .parseLenient()
+                              .append(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
+                              .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+                              .toFormatter())
+                          .toInstant(ZoneOffset.UTC).toEpochMilli() * 1000);
+                  break;
+                default:
+                  record.set(i, partition.getValues().get(i));
+                  break;
+              }
+            }
+            return ((Bound<Boolean>) boundExpression).eval(record);
+          }).collect(Collectors.toList());
+        }
       }
 
       return LegacyHiveTableUtils.toDirectoryInfos(partitions, current().spec());
@@ -199,7 +249,7 @@ private List<DirectoryInfo> getDirectoryInfosByFilter(Expression expression) {
   }
 
   private static DataFile createDataFile(FileStatus fileStatus, PartitionSpec partitionSpec, StructLike partitionData,
-      FileFormat format) {
+                                         FileFormat format) {
     DataFiles.Builder builder = DataFiles.builder(partitionSpec)
         .withPath(fileStatus.getPath().toString())
         .withFormat(format)

hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestHiveExpressions.java

@@ -36,7 +36,6 @@
 import static org.apache.iceberg.expressions.Expressions.notNull;
 import static org.apache.iceberg.expressions.Expressions.or;
 import static org.apache.iceberg.hive.legacy.HiveExpressions.simplifyPartitionFilter;
-import static org.apache.iceberg.hive.legacy.HiveExpressions.toPartitionFilterString;
 
 
 public class TestHiveExpressions {
@@ -118,10 +117,4 @@ public void testSimplifyRemoveNonPartitionColumnsWithinNot2() {
     Expression expected = alwaysTrue();
     Assert.assertEquals(expected.toString(), simplifyPartitionFilter(input, ImmutableSet.of("pcol")).toString());
   }
-
-  @Test
-  public void testToPartitionFilterStringEscapeStringLiterals() {
-    Expression input = equal("pcol", "s'1");
-    Assert.assertEquals("( pcol = 's\\'1' )", toPartitionFilterString(input));
-  }
 }