From 503755180b4f22f5e4dd484b607db4382f400f63 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Tue, 28 Apr 2020 19:08:42 +0800
Subject: [PATCH 01/10] Convert date and timestamp values in generics

---
 .../apache/iceberg/data/GenericRecord.java    | 45 +++++++++++++++--
 .../apache/iceberg/data/TestLocalScan.java    | 50 +++++++++++++++++--
 2 files changed, 87 insertions(+), 8 deletions(-)

diff --git a/data/src/main/java/org/apache/iceberg/data/GenericRecord.java b/data/src/main/java/org/apache/iceberg/data/GenericRecord.java
index e52e0b74259a..0c573c11a1cf 100644
--- a/data/src/main/java/org/apache/iceberg/data/GenericRecord.java
+++ b/data/src/main/java/org/apache/iceberg/data/GenericRecord.java
@@ -23,21 +23,32 @@
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Maps;
+import com.google.common.collect.BiMap;
+import com.google.common.collect.HashBiMap;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.StructType;
 
 public class GenericRecord implements Record, StructLike {
-  private static final LoadingCache<StructType, Map<String, Integer>> NAME_MAP_CACHE =
+  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
+  private static final LocalDate EPOCH_DAY = EPOCH.toLocalDate();
+  private static final LoadingCache<StructType, BiMap<String, Integer>> NAME_MAP_CACHE =
       Caffeine.newBuilder()
       .weakKeys()
       .build(struct -> {
-        Map<String, Integer> idToPos = Maps.newHashMap();
+        BiMap<String, Integer> idToPos = HashBiMap.create();
         List<Types.NestedField> fields = struct.fields();
         for (int i = 0; i < fields.size(); i += 1) {
           idToPos.put(fields.get(i).name(), i);
@@ -56,7 +67,7 @@ public static GenericRecord create(StructType struct) {
   private final StructType struct;
   private final int size;
   private final Object[] values;
-  private final Map<String, Integer> nameToPos;
+  private final BiMap<String, Integer> nameToPos;
 
   private GenericRecord(StructType struct) {
     this.struct = struct;
@@ -117,7 +128,31 @@ public Object get(int pos) {
   @Override
   public <T> T get(int pos, Class<T> javaClass) {
     Object value = get(pos);
-    if (value == null || javaClass.isInstance(value)) {
+
+    if (value == null) {
+      return javaClass.cast(null);
+    }
+
+    if (!javaClass.isInstance(value)) {
+      Type type = struct.field(nameToPos.inverse().get(pos)).type();
+      switch (type.typeId()) {
+        case TIMESTAMP:
+          if (((Types.TimestampType) type).shouldAdjustToUTC()) {
+            value = ChronoUnit.MICROS.between(EPOCH, (OffsetDateTime) value);
+          } else {
+            value = ChronoUnit.MICROS.between(EPOCH, ((LocalDateTime) value).atOffset(ZoneOffset.UTC));
+          }
+          break;
+        case DATE:
+          value = (int) ChronoUnit.DAYS.between(EPOCH_DAY, (LocalDate) value);
+          break;
+        case TIME:
+          value = ((LocalTime) value).toNanoOfDay() / 1000;
+          break;
+      }
+    }
+
+    if (javaClass.isInstance(value)) {
       return javaClass.cast(value);
     } else {
       throw new IllegalStateException("Not an instance of " + javaClass.getName() + ": " + value);
diff --git a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
index 673673097728..d50bb832a1ec 100644
--- a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
+++ b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
@@ -67,6 +67,7 @@
 import static com.google.common.collect.Iterables.filter;
 import static com.google.common.collect.Iterables.transform;
 import static org.apache.iceberg.DataFiles.fromInputFile;
+import static org.apache.iceberg.expressions.Expressions.equal;
 import static org.apache.iceberg.expressions.Expressions.lessThan;
 import static org.apache.iceberg.expressions.Expressions.lessThanOrEqual;
 import static org.apache.iceberg.hadoop.HadoopOutputFile.fromPath;
@@ -391,13 +392,17 @@ public void testAsOfTimeOlderThanFirstSnapshot() {
   }
 
   private DataFile writeFile(String location, String filename, List<Record> records) throws IOException {
+    return writeFile(location, filename, SCHEMA, records);
+  }
+
+  private DataFile writeFile(String location, String filename, Schema schema, List<Record> records) throws IOException {
     Path path = new Path(location, filename);
     FileFormat fileFormat = FileFormat.fromFileName(filename);
     Preconditions.checkNotNull(fileFormat, "Cannot determine format for file: %s", filename);
     switch (fileFormat) {
       case AVRO:
         FileAppender<Record> avroAppender = Avro.write(fromPath(path, CONF))
-            .schema(SCHEMA)
+            .schema(schema)
             .createWriterFunc(DataWriter::create)
             .named(fileFormat.name())
             .build();
@@ -414,7 +419,7 @@ private DataFile writeFile(String location, String filename, List<Record> record
 
       case PARQUET:
         FileAppender<Record> parquetAppender = Parquet.write(fromPath(path, CONF))
-            .schema(SCHEMA)
+            .schema(schema)
             .createWriterFunc(GenericParquetWriter::buildWriter)
             .build();
         try {
@@ -430,7 +435,7 @@ private DataFile writeFile(String location, String filename, List<Record> record
 
       case ORC:
         FileAppender<Record> orcAppender = ORC.write(fromPath(path, CONF))
-            .schema(SCHEMA)
+            .schema(schema)
             .createWriterFunc(GenericOrcWriter::buildWriter)
             .build();
         try {
@@ -449,6 +454,45 @@ private DataFile writeFile(String location, String filename, List<Record> record
     }
   }
 
+  @Test
+  public void testFilterWithComplexType() throws IOException {
+    if (format == FileFormat.PARQUET) {
+      return;
+    }
+
+    Schema schema = new Schema(
+        required(1, "timestamp_with_zone", Types.TimestampType.withZone()),
+        required(2, "timestamp_without_zone", Types.TimestampType.withoutZone()),
+        required(3, "date", Types.DateType.get()),
+        required(4, "time", Types.TimeType.get())
+    );
+
+    File tableLocation = temp.newFolder("complex_filter_table");
+    Assert.assertTrue(tableLocation.delete());
+
+    Table table = TABLES.create(
+        schema, PartitionSpec.unpartitioned(),
+        ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()),
+        tableLocation.getAbsolutePath());
+
+    List<Record> expected = RandomGenericData.generate(schema, 100, 435691832918L);
+    DataFile file = writeFile(tableLocation.toString(), format.addExtension("record-file"), schema, expected);
+    table.newFastAppend().appendFile(file).commit();
+
+    for (Record r : expected) {
+      Iterable<Record> filterResult = IcebergGenerics.read(table)
+          .where(equal("timestamp_with_zone", r.getField("timestamp_with_zone").toString()))
+          .where(equal("timestamp_without_zone", r.getField("timestamp_without_zone").toString()))
+          .where(equal("date", r.getField("date").toString()))
+          .where(equal("time", r.getField("time").toString()))
+          .build();
+
+      Assert.assertTrue(filterResult.iterator().hasNext());
+      Record readRecord = filterResult.iterator().next();
+      Assert.assertEquals(r.getField("timestamp_with_zone"), readRecord.getField("timestamp_with_zone"));
+    }
+  }
+
   private static ByteBuffer longToBuffer(long value) {
     return ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(0, value);
   }

From 174d1c8d2100cc24abfa8499fe37af3e584e20c0 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Wed, 29 Apr 2020 20:38:24 +0800
Subject: [PATCH 02/10] Add DateAccessor, TimeAccessor, and TimestampAccessor

---
 .../java/org/apache/iceberg/Accessors.java    | 62 ++++++++++++++++++-
 .../iceberg/transforms/TestResiduals.java     | 28 +++------
 .../apache/iceberg/data/GenericRecord.java    | 45 ++------------
 .../apache/iceberg/data/TestLocalScan.java    |  2 +-
 4 files changed, 74 insertions(+), 63 deletions(-)

diff --git a/api/src/main/java/org/apache/iceberg/Accessors.java b/api/src/main/java/org/apache/iceberg/Accessors.java
index 0e0d275c6440..0f366a540252 100644
--- a/api/src/main/java/org/apache/iceberg/Accessors.java
+++ b/api/src/main/java/org/apache/iceberg/Accessors.java
@@ -20,6 +20,13 @@
 package org.apache.iceberg;
 
 import com.google.common.collect.Maps;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.types.Type;
@@ -27,6 +34,9 @@
 import org.apache.iceberg.types.Types;
 
 public class Accessors {
+  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
+  private static final LocalDate EPOCH_DAY = EPOCH.toLocalDate();
+
   private Accessors() {
   }
 
@@ -170,7 +180,16 @@ public String toString() {
   }
 
   private static Accessor<StructLike> newAccessor(int pos, Type type) {
-    return new PositionAccessor(pos, type);
+    switch (type.typeId()) {
+      case DATE:
+        return new DateAccessor(pos, type);
+      case TIME:
+        return new TimeAccessor(pos, type);
+      case TIMESTAMP:
+        return new TimeStampAccessor(pos, type);
+      default:
+        return new PositionAccessor(pos, type);
+    }
   }
 
   private static Accessor<StructLike> newAccessor(int pos, boolean isOptional,
@@ -187,6 +206,47 @@ private static Accessor<StructLike> newAccessor(int pos, boolean isOptional,
     }
   }
 
+  private static class DateAccessor extends PositionAccessor {
+    private DateAccessor(int pos, Type type) {
+      super(pos, type);
+    }
+
+    @Override
+    public Object get(StructLike record) {
+      LocalDate value = record.get(position(), LocalDate.class);
+      return (int) ChronoUnit.DAYS.between(EPOCH_DAY, value);
+    }
+  }
+
+  private static class TimeAccessor extends PositionAccessor {
+    private TimeAccessor(int pos, Type type) {
+      super(pos, type);
+    }
+
+    @Override
+    public Object get(StructLike record) {
+      LocalTime value = record.get(position(), LocalTime.class);
+      return value.toNanoOfDay() / 1000;
+    }
+  }
+
+  private static class TimeStampAccessor extends PositionAccessor {
+    private TimeStampAccessor(int pos, Type type) {
+      super(pos, type);
+    }
+
+    @Override
+    public Object get(StructLike record) {
+      if (((Types.TimestampType) type().asPrimitiveType()).shouldAdjustToUTC()) {
+        OffsetDateTime value = record.get(position(), OffsetDateTime.class);
+        return ChronoUnit.MICROS.between(EPOCH, value);
+      } else {
+        LocalDateTime value = record.get(position(), LocalDateTime.class);
+        return ChronoUnit.MICROS.between(EPOCH, value.atOffset(ZoneOffset.UTC));
+      }
+    }
+  }
+
   private static class BuildPositionAccessors extends TypeUtil.SchemaVisitor<Map<Integer, Accessor<StructLike>>> {
 
     @Override
diff --git a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
index 2264800b9117..380667836993 100644
--- a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
+++ b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
@@ -19,13 +19,13 @@
 
 package org.apache.iceberg.transforms;
 
+import java.time.LocalDate;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.TestHelpers.Row;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.expressions.Predicate;
 import org.apache.iceberg.expressions.ResidualEvaluator;
 import org.apache.iceberg.expressions.UnboundPredicate;
@@ -191,26 +191,18 @@ public void testInTimestamp() {
         Types.NestedField.optional(51, "dateint", Types.IntegerType.get())
     );
 
-    Long date20191201 = (Long) Literal.of("2019-12-01T00:00:00.00000")
-        .to(Types.TimestampType.withoutZone()).value();
-    Long date20191202 = (Long) Literal.of("2019-12-02T00:00:00.00000")
-        .to(Types.TimestampType.withoutZone()).value();
-
     PartitionSpec spec = PartitionSpec.builderFor(schema)
         .day("ts")
         .build();
 
-    Transform day = spec.getFieldsBySourceId(50).get(0).transform();
-    Integer tsDay = (Integer) day.apply(date20191201);
-
-    Predicate pred = in("ts", date20191201, date20191202);
+    Predicate pred = in("ts", "2019-12-01T00:00:00.00000", "2019-12-01T00:00:00.00000");
     ResidualEvaluator resEval = ResidualEvaluator.of(spec,
         pred, true);
 
-    Expression residual = resEval.residualFor(Row.of(tsDay));
+    Expression residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01")));
     Assert.assertEquals("Residual should be the original in predicate", pred, residual);
 
-    residual = resEval.residualFor(Row.of(tsDay + 3));
+    residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01").plusDays(3)));
     Assert.assertEquals("Residual should be alwaysFalse", alwaysFalse(), residual);
   }
 
@@ -242,26 +234,20 @@ public void testNotInTimestamp() {
         Types.NestedField.optional(51, "dateint", Types.IntegerType.get())
     );
 
-    Long date20191201 = (Long) Literal.of("2019-12-01T00:00:00.00000")
-        .to(Types.TimestampType.withoutZone()).value();
-    Long date20191202 = (Long) Literal.of("2019-12-02T00:00:00.00000")
-        .to(Types.TimestampType.withoutZone()).value();
-
     PartitionSpec spec = PartitionSpec.builderFor(schema)
         .day("ts")
         .build();
 
     Transform day = spec.getFieldsBySourceId(50).get(0).transform();
-    Integer tsDay = (Integer) day.apply(date20191201);
 
-    Predicate pred = notIn("ts", date20191201, date20191202);
+    Predicate pred = notIn("ts", "2019-12-01T00:00:00.00000", "2019-12-02T00:00:00.00000");
     ResidualEvaluator resEval = ResidualEvaluator.of(spec,
         pred, true);
 
-    Expression residual = resEval.residualFor(Row.of(tsDay));
+    Expression residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01")));
     Assert.assertEquals("Residual should be the original notIn predicate", pred, residual);
 
-    residual = resEval.residualFor(Row.of(tsDay + 3));
+    residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01").plusDays(3)));
     Assert.assertEquals("Residual should be alwaysTrue", alwaysTrue(), residual);
   }
 }
diff --git a/data/src/main/java/org/apache/iceberg/data/GenericRecord.java b/data/src/main/java/org/apache/iceberg/data/GenericRecord.java
index 0c573c11a1cf..e52e0b74259a 100644
--- a/data/src/main/java/org/apache/iceberg/data/GenericRecord.java
+++ b/data/src/main/java/org/apache/iceberg/data/GenericRecord.java
@@ -23,32 +23,21 @@
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.BiMap;
-import com.google.common.collect.HashBiMap;
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
-import java.time.temporal.ChronoUnit;
+import com.google.common.collect.Maps;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
-import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.StructType;
 
 public class GenericRecord implements Record, StructLike {
-  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
-  private static final LocalDate EPOCH_DAY = EPOCH.toLocalDate();
-  private static final LoadingCache<StructType, BiMap<String, Integer>> NAME_MAP_CACHE =
+  private static final LoadingCache<StructType, Map<String, Integer>> NAME_MAP_CACHE =
       Caffeine.newBuilder()
       .weakKeys()
       .build(struct -> {
-        BiMap<String, Integer> idToPos = HashBiMap.create();
+        Map<String, Integer> idToPos = Maps.newHashMap();
         List<Types.NestedField> fields = struct.fields();
         for (int i = 0; i < fields.size(); i += 1) {
           idToPos.put(fields.get(i).name(), i);
@@ -67,7 +56,7 @@ public static GenericRecord create(StructType struct) {
   private final StructType struct;
   private final int size;
   private final Object[] values;
-  private final BiMap<String, Integer> nameToPos;
+  private final Map<String, Integer> nameToPos;
 
   private GenericRecord(StructType struct) {
     this.struct = struct;
@@ -128,31 +117,7 @@ public Object get(int pos) {
   @Override
   public <T> T get(int pos, Class<T> javaClass) {
     Object value = get(pos);
-
-    if (value == null) {
-      return javaClass.cast(null);
-    }
-
-    if (!javaClass.isInstance(value)) {
-      Type type = struct.field(nameToPos.inverse().get(pos)).type();
-      switch (type.typeId()) {
-        case TIMESTAMP:
-          if (((Types.TimestampType) type).shouldAdjustToUTC()) {
-            value = ChronoUnit.MICROS.between(EPOCH, (OffsetDateTime) value);
-          } else {
-            value = ChronoUnit.MICROS.between(EPOCH, ((LocalDateTime) value).atOffset(ZoneOffset.UTC));
-          }
-          break;
-        case DATE:
-          value = (int) ChronoUnit.DAYS.between(EPOCH_DAY, (LocalDate) value);
-          break;
-        case TIME:
-          value = ((LocalTime) value).toNanoOfDay() / 1000;
-          break;
-      }
-    }
-
-    if (javaClass.isInstance(value)) {
+    if (value == null || javaClass.isInstance(value)) {
       return javaClass.cast(value);
     } else {
       throw new IllegalStateException("Not an instance of " + javaClass.getName() + ": " + value);
diff --git a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
index d50bb832a1ec..2159a3101416 100644
--- a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
+++ b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
@@ -455,7 +455,7 @@ private DataFile writeFile(String location, String filename, Schema schema, List
   }
 
   @Test
-  public void testFilterWithComplexType() throws IOException {
+  public void testFilterWithDateAndTimestamp() throws IOException {
     if (format == FileFormat.PARQUET) {
       return;
     }

From 2f7e29e83fe4d5791b28c0953176129d63efa75d Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Tue, 5 May 2020 21:12:34 +0800
Subject: [PATCH 03/10] revert unit test changes

---
 .../iceberg/transforms/TestResiduals.java     | 27 ++++++++++++++-----
 1 file changed, 21 insertions(+), 6 deletions(-)

diff --git a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
index 380667836993..ed077353047f 100644
--- a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
+++ b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
@@ -26,6 +26,7 @@
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.expressions.Predicate;
 import org.apache.iceberg.expressions.ResidualEvaluator;
 import org.apache.iceberg.expressions.UnboundPredicate;
@@ -191,18 +192,26 @@ public void testInTimestamp() {
         Types.NestedField.optional(51, "dateint", Types.IntegerType.get())
     );
 
+    Long date20191201 = (Long) Literal.of("2019-12-01T00:00:00.00000")
+        .to(Types.TimestampType.withoutZone()).value();
+    Long date20191202 = (Long) Literal.of("2019-12-02T00:00:00.00000")
+        .to(Types.TimestampType.withoutZone()).value();
+
     PartitionSpec spec = PartitionSpec.builderFor(schema)
         .day("ts")
         .build();
 
-    Predicate pred = in("ts", "2019-12-01T00:00:00.00000", "2019-12-01T00:00:00.00000");
+    Transform day = spec.getFieldsBySourceId(50).get(0).transform();
+    Integer tsDay = (Integer) day.apply(date20191201);
+
+    Predicate pred = in("ts", date20191201, date20191202);
     ResidualEvaluator resEval = ResidualEvaluator.of(spec,
         pred, true);
 
-    Expression residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01")));
+    Expression residual = resEval.residualFor(Row.of(tsDay));
     Assert.assertEquals("Residual should be the original in predicate", pred, residual);
 
-    residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01").plusDays(3)));
+    residual = resEval.residualFor(Row.of(tsDay + 3));
     Assert.assertEquals("Residual should be alwaysFalse", alwaysFalse(), residual);
   }
 
@@ -234,20 +243,26 @@ public void testNotInTimestamp() {
         Types.NestedField.optional(51, "dateint", Types.IntegerType.get())
     );
 
+    Long date20191201 = (Long) Literal.of("2019-12-01T00:00:00.00000")
+        .to(Types.TimestampType.withoutZone()).value();
+    Long date20191202 = (Long) Literal.of("2019-12-02T00:00:00.00000")
+        .to(Types.TimestampType.withoutZone()).value();
+
     PartitionSpec spec = PartitionSpec.builderFor(schema)
         .day("ts")
         .build();
 
     Transform day = spec.getFieldsBySourceId(50).get(0).transform();
+    Integer tsDay = (Integer) day.apply(date20191201);
 
-    Predicate pred = notIn("ts", "2019-12-01T00:00:00.00000", "2019-12-02T00:00:00.00000");
+    Predicate pred = notIn("ts", date20191201, date20191202);
     ResidualEvaluator resEval = ResidualEvaluator.of(spec,
         pred, true);
 
-    Expression residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01")));
+    Expression residual = resEval.residualFor(Row.of(tsDay));
     Assert.assertEquals("Residual should be the original notIn predicate", pred, residual);
 
-    residual = resEval.residualFor(Row.of(LocalDate.parse("2019-12-01").plusDays(3)));
+    residual = resEval.residualFor(Row.of(tsDay + 3));
     Assert.assertEquals("Residual should be alwaysTrue", alwaysTrue(), residual);
   }
 }

From d7f3d5000df96bfa5d2d989e73a2edcc7d0f36b6 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Sat, 9 May 2020 20:42:34 +0800
Subject: [PATCH 04/10] rebase and remove parquet work around

---
 .../java/org/apache/iceberg/transforms/TestResiduals.java     | 1 -
 data/src/main/java/org/apache/iceberg/data/Accessors.java     | 4 ++++
 data/src/test/java/org/apache/iceberg/data/TestLocalScan.java | 4 ----
 3 files changed, 4 insertions(+), 5 deletions(-)
 create mode 100644 data/src/main/java/org/apache/iceberg/data/Accessors.java

diff --git a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
index ed077353047f..2264800b9117 100644
--- a/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
+++ b/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java
@@ -19,7 +19,6 @@
 
 package org.apache.iceberg.transforms;
 
-import java.time.LocalDate;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.TestHelpers.Row;
diff --git a/data/src/main/java/org/apache/iceberg/data/Accessors.java b/data/src/main/java/org/apache/iceberg/data/Accessors.java
new file mode 100644
index 000000000000..faf8c43fd289
--- /dev/null
+++ b/data/src/main/java/org/apache/iceberg/data/Accessors.java
@@ -0,0 +1,4 @@
+package org.apache.iceberg.data;
+
+public class Accessors {
+}
diff --git a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
index 2159a3101416..c15f59fc202e 100644
--- a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
+++ b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
@@ -456,10 +456,6 @@ private DataFile writeFile(String location, String filename, Schema schema, List
 
   @Test
   public void testFilterWithDateAndTimestamp() throws IOException {
-    if (format == FileFormat.PARQUET) {
-      return;
-    }
-
     Schema schema = new Schema(
         required(1, "timestamp_with_zone", Types.TimestampType.withZone()),
         required(2, "timestamp_without_zone", Types.TimestampType.withoutZone()),

From ea42ccec9207ba69ea0d3fd0bccc73c142fb6207 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Sat, 9 May 2020 21:34:29 +0800
Subject: [PATCH 05/10] alternative fix

---
 .../java/org/apache/iceberg/Accessors.java    |  11 +-
 .../apache/iceberg/expressions/Binder.java    |  40 ++-
 .../apache/iceberg/expressions/Evaluator.java |   7 +
 .../iceberg/expressions/NamedReference.java   |  15 +-
 .../apache/iceberg/expressions/Unbound.java   |  13 +
 .../iceberg/expressions/UnboundPredicate.java |  18 ++
 .../iceberg/expressions/UnboundTransform.java |  12 +-
 .../org/apache/iceberg/data/Accessors.java    | 281 ++++++++++++++++++
 .../apache/iceberg/data/GenericEvaluator.java |  34 +++
 .../iceberg/data/TableScanIterable.java       |   6 +-
 10 files changed, 423 insertions(+), 14 deletions(-)
 create mode 100644 data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java

diff --git a/api/src/main/java/org/apache/iceberg/Accessors.java b/api/src/main/java/org/apache/iceberg/Accessors.java
index 0f366a540252..591494ed8473 100644
--- a/api/src/main/java/org/apache/iceberg/Accessors.java
+++ b/api/src/main/java/org/apache/iceberg/Accessors.java
@@ -180,16 +180,7 @@ public String toString() {
   }
 
   private static Accessor<StructLike> newAccessor(int pos, Type type) {
-    switch (type.typeId()) {
-      case DATE:
-        return new DateAccessor(pos, type);
-      case TIME:
-        return new TimeAccessor(pos, type);
-      case TIMESTAMP:
-        return new TimeStampAccessor(pos, type);
-      default:
-        return new PositionAccessor(pos, type);
-    }
+    return new PositionAccessor(pos, type);
   }
 
   private static Accessor<StructLike> newAccessor(int pos, boolean isOptional,
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Binder.java b/api/src/main/java/org/apache/iceberg/expressions/Binder.java
index 1bc6563b0bc4..23196a2a6baf 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Binder.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Binder.java
@@ -22,7 +22,10 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.ExpressionVisitors.ExpressionVisitor;
 import org.apache.iceberg.types.Type;
@@ -62,6 +65,34 @@ public static Expression bind(StructType struct,
     return ExpressionVisitors.visit(expr, new BindVisitor(struct, caseSensitive));
   }
 
+  /**
+   * Replaces all unbound/named references with bound references to fields in the given struct.
+   * <p>
+   * When a reference is resolved, any literal used in a predicate for that field is converted to
+   * the field's type using {@link Literal#to(Type)}. If automatic conversion to that type isn't
+   * allowed, a {@link ValidationException validation exception} is thrown.
+   * <p>
+   * The result expression may be simplified when constructed. For example, {@code isNull("a")} is
+   * replaced with {@code alwaysFalse()} when {@code "a"} is resolved to a required field.
+   * <p>
+   * The expression cannot contain references that are already bound, or an
+   * {@link IllegalStateException} will be thrown.
+   *
+   * @param struct The {@link StructType struct type} to resolve references by name.
+   * @param expr An {@link Expression expression} to rewrite with bound references.
+   * @param accessors A map to {@link Accessor accessor} to access the {@link StructLike StructLike} value
+   * @param caseSensitive A boolean flag to control whether the bind should enforce case sensitivity.
+   * @return the expression rewritten with bound references
+   * @throws ValidationException if literals do not match bound references
+   * @throws IllegalStateException if any references are already bound
+   */
+  public static Expression bind(StructType struct,
+                                Expression expr,
+                                Map<Integer, Accessor<StructLike>> accessors,
+                                boolean caseSensitive) {
+    return ExpressionVisitors.visit(expr, new BindVisitor(struct, accessors, caseSensitive));
+  }
+
   /**
    * Replaces all unbound/named references with bound references to fields in the given struct,
    * defaulting to case sensitive mode.
@@ -104,12 +135,19 @@ public static Set<Integer> boundReferences(StructType struct, List<Expression> e
   private static class BindVisitor extends ExpressionVisitor<Expression> {
     private final StructType struct;
     private final boolean caseSensitive;
+    private Map<Integer, Accessor<StructLike>> accessors = null;
 
     private BindVisitor(StructType struct, boolean caseSensitive) {
       this.struct = struct;
       this.caseSensitive = caseSensitive;
     }
 
+    private BindVisitor(StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive) {
+      this.struct = struct;
+      this.caseSensitive = caseSensitive;
+      this.accessors = accessors;
+    }
+
     @Override
     public Expression alwaysTrue() {
       return Expressions.alwaysTrue();
@@ -142,7 +180,7 @@ public <T> Expression predicate(BoundPredicate<T> pred) {
 
     @Override
     public <T> Expression predicate(UnboundPredicate<T> pred) {
-      return pred.bind(struct, caseSensitive);
+      return accessors == null ? pred.bind(struct, caseSensitive) : pred.bind(struct, accessors, caseSensitive);
     }
   }
 
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
index 0d3e1ef311f8..878b73d0703c 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
@@ -21,7 +21,9 @@
 
 import java.io.Serializable;
 import java.util.Comparator;
+import java.util.Map;
 import java.util.Set;
+import org.apache.iceberg.Accessor;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.expressions.ExpressionVisitors.BoundVisitor;
 import org.apache.iceberg.types.Types.StructType;
@@ -52,6 +54,11 @@ public Evaluator(StructType struct, Expression unbound, boolean caseSensitive) {
     this.expr = Binder.bind(struct, unbound, caseSensitive);
   }
 
+  public Evaluator(StructType struct, Expression unbound, Map<Integer, Accessor<StructLike>> accessors,
+                   boolean caseSensitive) {
+    this.expr = Binder.bind(struct, unbound, accessors, caseSensitive);
+  }
+
   public boolean eval(StructLike data) {
     return visitor().eval(data);
   }
diff --git a/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java b/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
index d928b3abfe1d..552f1dbef34f 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
@@ -20,7 +20,10 @@
 package org.apache.iceberg.expressions;
 
 import com.google.common.base.Preconditions;
+import java.util.Map;
+import org.apache.iceberg.Accessor;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.types.Types;
 
@@ -38,6 +41,12 @@ public String name() {
 
   @Override
   public BoundReference<T> bind(Types.StructType struct, boolean caseSensitive) {
+    return bind(struct, null, caseSensitive);
+  }
+
+  @Override
+  public BoundReference<T> bind(Types.StructType struct, Map<Integer, Accessor<StructLike>> accessors,
+                                boolean caseSensitive) {
     Schema schema = new Schema(struct.fields());
     Types.NestedField field = caseSensitive ?
         schema.findField(name) :
@@ -46,7 +55,11 @@ public BoundReference<T> bind(Types.StructType struct, boolean caseSensitive) {
     ValidationException.check(field != null,
         "Cannot find field '%s' in struct: %s", name, schema.asStruct());
 
-    return new BoundReference<>(field, schema.accessorForField(field.fieldId()));
+    if (accessors != null) {
+      return new BoundReference<>(field, accessors.get(field.fieldId()));
+    } else {
+      return new BoundReference<>(field, schema.accessorForField(field.fieldId()));
+    }
   }
 
   @Override
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Unbound.java b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
index f160f935fb3e..17a4a05244fd 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
@@ -19,6 +19,9 @@
 
 package org.apache.iceberg.expressions;
 
+import java.util.Map;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.types.Types;
 
 /**
@@ -36,6 +39,16 @@ public interface Unbound<T, B> {
    */
   B bind(Types.StructType struct, boolean caseSensitive);
 
+  /**
+   * Bind this value expression to concrete types.
+   *
+   * @param struct input data types
+   * @param accessors the accessors for the struct
+   * @param caseSensitive whether binding should match columns using case sensitive resolution
+   * @return a bound value expression
+   */
+  B bind(Types.StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive);
+
   /**
    * @return this expression's underlying reference
    */
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
index fbef5e158934..fa4af9e2b549 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
@@ -25,7 +25,10 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.CharSequenceSet;
@@ -115,6 +118,21 @@ public Expression bind(StructType struct, boolean caseSensitive) {
     return bindLiteralOperation(bound);
   }
 
+  @Override
+  public Expression bind(StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive) {
+    BoundTerm<T> bound = term().bind(struct, accessors, caseSensitive);
+
+    if (literals == null) {
+      return bindUnaryOperation(bound);
+    }
+
+    if (op() == Operation.IN || op() == Operation.NOT_IN) {
+      return bindInOperation(bound);
+    }
+
+    return bindLiteralOperation(bound);
+  }
+
   private Expression bindUnaryOperation(BoundTerm<T> boundTerm) {
     switch (op()) {
       case IS_NULL:
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
index be3cec188935..ea0ff9205f04 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
@@ -19,6 +19,9 @@
 
 package org.apache.iceberg.expressions;
 
+import java.util.Map;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
@@ -45,7 +48,14 @@ public Transform<S, T> transform() {
   @SuppressWarnings("unchecked")
   @Override
   public BoundTransform<S, T> bind(Types.StructType struct, boolean caseSensitive) {
-    BoundReference<S> boundRef = ref.bind(struct, caseSensitive);
+    return bind(struct, null, caseSensitive);
+  }
+
+  @Override
+  public BoundTransform<S, T> bind(Types.StructType struct, Map<Integer, Accessor<StructLike>> accessors,
+                                   boolean caseSensitive) {
+    BoundReference<S> boundRef =
+        accessors == null ? ref.bind(struct, caseSensitive) : ref.bind(struct, accessors, caseSensitive);
 
     Transform<S, T> typeTransform;
     try {
diff --git a/data/src/main/java/org/apache/iceberg/data/Accessors.java b/data/src/main/java/org/apache/iceberg/data/Accessors.java
index faf8c43fd289..39852884c7f7 100644
--- a/data/src/main/java/org/apache/iceberg/data/Accessors.java
+++ b/data/src/main/java/org/apache/iceberg/data/Accessors.java
@@ -1,4 +1,285 @@
+/*
+ * 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.data;
 
+import com.google.common.collect.Maps;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+
 public class Accessors {
+  private Accessors() {
+  }
+
+  public static Integer toPosition(Accessor<StructLike> accessor) {
+    if (accessor instanceof PositionAccessor) {
+      return ((PositionAccessor) accessor).position();
+    }
+    throw new IllegalArgumentException("Cannot convert nested accessor to position");
+  }
+
+  static Map<Integer, Accessor<StructLike>> forSchema(Schema schema) {
+    return TypeUtil.visit(schema, new BuildPositionAccessors());
+  }
+
+  private static class PositionAccessor implements Accessor<StructLike> {
+    private int position;
+    private final Type type;
+    private final Class<?> javaClass;
+
+    PositionAccessor(int pos, Type type) {
+      this.position = pos;
+      this.type = type;
+      this.javaClass = type.typeId().javaClass();
+    }
+
+    @Override
+    public Object get(StructLike row) {
+      return row.get(position, javaClass);
+    }
+
+    @Override
+    public Type type() {
+      return type;
+    }
+
+    public int position() {
+      return position;
+    }
+
+    public Class<?> javaClass() {
+      return javaClass;
+    }
+
+    @Override
+    public String toString() {
+      return "Accessor(positions=[" + position + "], type=" + type + ")";
+    }
+  }
+
+  private static class Position2Accessor implements Accessor<StructLike> {
+    private final int p0;
+    private final int p1;
+    private final Type type;
+    private final Class<?> javaClass;
+
+    Position2Accessor(int pos, PositionAccessor wrapped) {
+      this.p0 = pos;
+      this.p1 = wrapped.position();
+      this.type = wrapped.type();
+      this.javaClass = wrapped.javaClass();
+    }
+
+    @Override
+    public Object get(StructLike row) {
+      return row.get(p0, StructLike.class).get(p1, javaClass);
+    }
+
+    @Override
+    public Type type() {
+      return type;
+    }
+
+    public Class<?> javaClass() {
+      return javaClass;
+    }
+
+    @Override
+    public String toString() {
+      return "Accessor(positions=[" + p0 + ", " + p1 + "], type=" + type + ")";
+    }
+  }
+
+  private static class Position3Accessor implements Accessor<StructLike> {
+    private final int p0;
+    private final int p1;
+    private final int p2;
+    private final Type type;
+    private final Class<?> javaClass;
+
+    Position3Accessor(int pos, Position2Accessor wrapped) {
+      this.p0 = pos;
+      this.p1 = wrapped.p0;
+      this.p2 = wrapped.p1;
+      this.type = wrapped.type();
+      this.javaClass = wrapped.javaClass();
+    }
+
+    @Override
+    public Object get(StructLike row) {
+      return row.get(p0, StructLike.class).get(p1, StructLike.class).get(p2, javaClass);
+    }
+
+    @Override
+    public Type type() {
+      return type;
+    }
+
+    @Override
+    public String toString() {
+      return "Accessor(positions=[" + p0 + ", " + p1 + ", " + p2 + "], type=" + type + ")";
+    }
+  }
+
+  private static class WrappedPositionAccessor implements Accessor<StructLike> {
+    private final int position;
+    private final Accessor<StructLike> accessor;
+
+    WrappedPositionAccessor(int pos, Accessor<StructLike> accessor) {
+      this.position = pos;
+      this.accessor = accessor;
+    }
+
+    @Override
+    public Object get(StructLike row) {
+      StructLike inner = row.get(position, StructLike.class);
+      if (inner != null) {
+        return accessor.get(inner);
+      }
+      return null;
+    }
+
+    @Override
+    public Type type() {
+      return accessor.type();
+    }
+
+    @Override
+    public String toString() {
+      return "WrappedAccessor(position=" + position + ", wrapped=" + accessor + ")";
+    }
+  }
+
+  private static Accessor<StructLike> newAccessor(int pos, Type type) {
+    switch (type.typeId()) {
+      case DATE:
+        return new DateAccessor(pos, type);
+      case TIME:
+        return new TimeAccessor(pos, type);
+      case TIMESTAMP:
+        return new TimeStampAccessor(pos, type);
+      default:
+        return new PositionAccessor(pos, type);
+    }
+  }
+
+  private static Accessor<StructLike> newAccessor(int pos, boolean isOptional,
+                                                  Accessor<StructLike> accessor) {
+    if (isOptional) {
+      // the wrapped position handles null layers
+      return new WrappedPositionAccessor(pos, accessor);
+    } else if (accessor instanceof PositionAccessor) {
+      return new Position2Accessor(pos, (PositionAccessor) accessor);
+    } else if (accessor instanceof Position2Accessor) {
+      return new Position3Accessor(pos, (Position2Accessor) accessor);
+    } else {
+      return new WrappedPositionAccessor(pos, accessor);
+    }
+  }
+
+  private static class DateAccessor extends PositionAccessor {
+    private DateAccessor(int pos, Type type) {
+      super(pos, type);
+    }
+
+    @Override
+    public Object get(StructLike record) {
+      LocalDate value = record.get(position(), LocalDate.class);
+      return DateTimeUtil.daysFromDate(value);
+    }
+  }
+
+  private static class TimeAccessor extends PositionAccessor {
+    private TimeAccessor(int pos, Type type) {
+      super(pos, type);
+    }
+
+    @Override
+    public Object get(StructLike record) {
+      LocalTime value = record.get(position(), LocalTime.class);
+      return DateTimeUtil.microsFromTime(value);
+    }
+  }
+
+  private static class TimeStampAccessor extends PositionAccessor {
+    private TimeStampAccessor(int pos, Type type) {
+      super(pos, type);
+    }
+
+    @Override
+    public Object get(StructLike record) {
+      if (((Types.TimestampType) type().asPrimitiveType()).shouldAdjustToUTC()) {
+        OffsetDateTime value = record.get(position(), OffsetDateTime.class);
+        return DateTimeUtil.microsFromTimestamptz(value);
+      } else {
+        LocalDateTime value = record.get(position(), LocalDateTime.class);
+        return DateTimeUtil.microsFromTimestamp(value);
+      }
+    }
+  }
+
+  private static class BuildPositionAccessors extends TypeUtil.SchemaVisitor<Map<Integer, Accessor<StructLike>>> {
+
+    @Override
+    public Map<Integer, Accessor<StructLike>> schema(
+        Schema schema, Map<Integer, Accessor<StructLike>> structResult) {
+      return structResult;
+    }
+
+    @Override
+    public Map<Integer, Accessor<StructLike>> struct(
+        Types.StructType struct, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
+      Map<Integer, Accessor<StructLike>> accessors = Maps.newHashMap();
+      List<Types.NestedField> fields = struct.fields();
+      for (int i = 0; i < fieldResults.size(); i += 1) {
+        Types.NestedField field = fields.get(i);
+        Map<Integer, Accessor<StructLike>> result = fieldResults.get(i);
+        if (result != null) {
+          for (Map.Entry<Integer, Accessor<StructLike>> entry : result.entrySet()) {
+            accessors.put(entry.getKey(), newAccessor(i, field.isOptional(), entry.getValue()));
+          }
+        } else {
+          accessors.put(field.fieldId(), newAccessor(i, field.type()));
+        }
+      }
+
+      if (accessors.isEmpty()) {
+        return null;
+      }
+
+      return accessors;
+    }
+
+    @Override
+    public Map<Integer, Accessor<StructLike>> field(
+        Types.NestedField field, Map<Integer, Accessor<StructLike>> fieldResult) {
+      return fieldResult;
+    }
+  }
 }
diff --git a/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java b/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
new file mode 100644
index 000000000000..88c078597ba7
--- /dev/null
+++ b/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
@@ -0,0 +1,34 @@
+/*
+ * 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.data;
+
+import java.util.Map;
+import org.apache.iceberg.Accessor;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.types.Types;
+
+public class GenericEvaluator extends Evaluator {
+  public GenericEvaluator(Types.StructType structType, Expression unbound, Map<Integer, Accessor<StructLike>> accessors,
+                          boolean caseSensitive) {
+    super(structType, unbound, accessors, caseSensitive);
+  }
+}
diff --git a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
index a683c019844f..9056aa3c9683 100644
--- a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
+++ b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
@@ -29,10 +29,12 @@
 import java.util.Map;
 import java.util.NoSuchElementException;
 import org.apache.avro.generic.GenericData;
+import org.apache.iceberg.Accessor;
 import org.apache.iceberg.CombinedScanTask;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.avro.Avro;
@@ -160,7 +162,9 @@ public boolean hasNext() {
           this.currentCloseable = reader;
 
           if (task.residual() != null && task.residual() != Expressions.alwaysTrue()) {
-            Evaluator filter = new Evaluator(projection.asStruct(), task.residual(), caseSensitive);
+            Map<Integer, Accessor<StructLike>> accessors = Accessors.forSchema(projection);
+            Evaluator filter = new GenericEvaluator(projection.asStruct(), task.residual(),
+                accessors, caseSensitive);
             this.currentIterator = Iterables.filter(reader, filter::eval).iterator();
           } else {
             this.currentIterator = reader.iterator();

From f25ba39d91c5e4d3dd21c78a000cb1533f8545de Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Sun, 10 May 2020 14:45:08 +0800
Subject: [PATCH 06/10] Add comments and remove useless chagnes

---
 .../java/org/apache/iceberg/Accessors.java    | 51 -------------------
 .../apache/iceberg/expressions/Unbound.java   |  2 +-
 .../iceberg/expressions/UnboundPredicate.java | 24 ++++-----
 .../org/apache/iceberg/data/Accessors.java    |  7 ---
 4 files changed, 13 insertions(+), 71 deletions(-)

diff --git a/api/src/main/java/org/apache/iceberg/Accessors.java b/api/src/main/java/org/apache/iceberg/Accessors.java
index 591494ed8473..0e0d275c6440 100644
--- a/api/src/main/java/org/apache/iceberg/Accessors.java
+++ b/api/src/main/java/org/apache/iceberg/Accessors.java
@@ -20,13 +20,6 @@
 package org.apache.iceberg;
 
 import com.google.common.collect.Maps;
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
-import java.time.temporal.ChronoUnit;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.types.Type;
@@ -34,9 +27,6 @@
 import org.apache.iceberg.types.Types;
 
 public class Accessors {
-  private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
-  private static final LocalDate EPOCH_DAY = EPOCH.toLocalDate();
-
   private Accessors() {
   }
 
@@ -197,47 +187,6 @@ private static Accessor<StructLike> newAccessor(int pos, boolean isOptional,
     }
   }
 
-  private static class DateAccessor extends PositionAccessor {
-    private DateAccessor(int pos, Type type) {
-      super(pos, type);
-    }
-
-    @Override
-    public Object get(StructLike record) {
-      LocalDate value = record.get(position(), LocalDate.class);
-      return (int) ChronoUnit.DAYS.between(EPOCH_DAY, value);
-    }
-  }
-
-  private static class TimeAccessor extends PositionAccessor {
-    private TimeAccessor(int pos, Type type) {
-      super(pos, type);
-    }
-
-    @Override
-    public Object get(StructLike record) {
-      LocalTime value = record.get(position(), LocalTime.class);
-      return value.toNanoOfDay() / 1000;
-    }
-  }
-
-  private static class TimeStampAccessor extends PositionAccessor {
-    private TimeStampAccessor(int pos, Type type) {
-      super(pos, type);
-    }
-
-    @Override
-    public Object get(StructLike record) {
-      if (((Types.TimestampType) type().asPrimitiveType()).shouldAdjustToUTC()) {
-        OffsetDateTime value = record.get(position(), OffsetDateTime.class);
-        return ChronoUnit.MICROS.between(EPOCH, value);
-      } else {
-        LocalDateTime value = record.get(position(), LocalDateTime.class);
-        return ChronoUnit.MICROS.between(EPOCH, value.atOffset(ZoneOffset.UTC));
-      }
-    }
-  }
-
   private static class BuildPositionAccessors extends TypeUtil.SchemaVisitor<Map<Integer, Accessor<StructLike>>> {
 
     @Override
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Unbound.java b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
index 17a4a05244fd..ac50e3b1949b 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
@@ -43,7 +43,7 @@ public interface Unbound<T, B> {
    * Bind this value expression to concrete types.
    *
    * @param struct input data types
-   * @param accessors the accessors for the struct
+   * @param accessors the accessors for the struct like value
    * @param caseSensitive whether binding should match columns using case sensitive resolution
    * @return a bound value expression
    */
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
index fa4af9e2b549..5a0400c514a2 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
@@ -105,22 +105,22 @@ Expression bind(StructType struct) {
    */
   @Override
   public Expression bind(StructType struct, boolean caseSensitive) {
-    BoundTerm<T> bound = term().bind(struct, caseSensitive);
-
-    if (literals == null) {
-      return bindUnaryOperation(bound);
-    }
-
-    if (op() == Operation.IN || op() == Operation.NOT_IN) {
-      return bindInOperation(bound);
-    }
-
-    return bindLiteralOperation(bound);
+    return bind(struct, null, caseSensitive);
   }
 
+  /**
+   * Bind this UnboundPredicate.
+   *
+   * @param struct The {@link StructType struct type} to resolve references by name.
+   * @param accessors The accessors used to access values in struct
+   * @param caseSensitive A boolean flag to control whether the bind should enforce case sensitivity.
+   * @return an {@link Expression}
+   * @throws ValidationException if literals do not match bound references, or if comparison on expression is invalid
+   */
   @Override
   public Expression bind(StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive) {
-    BoundTerm<T> bound = term().bind(struct, accessors, caseSensitive);
+    BoundTerm<T> bound =
+        accessors == null ? term().bind(struct, caseSensitive) : term().bind(struct, accessors, caseSensitive);
 
     if (literals == null) {
       return bindUnaryOperation(bound);
diff --git a/data/src/main/java/org/apache/iceberg/data/Accessors.java b/data/src/main/java/org/apache/iceberg/data/Accessors.java
index 39852884c7f7..4172c4d14da2 100644
--- a/data/src/main/java/org/apache/iceberg/data/Accessors.java
+++ b/data/src/main/java/org/apache/iceberg/data/Accessors.java
@@ -37,13 +37,6 @@ public class Accessors {
   private Accessors() {
   }
 
-  public static Integer toPosition(Accessor<StructLike> accessor) {
-    if (accessor instanceof PositionAccessor) {
-      return ((PositionAccessor) accessor).position();
-    }
-    throw new IllegalArgumentException("Cannot convert nested accessor to position");
-  }
-
   static Map<Integer, Accessor<StructLike>> forSchema(Schema schema) {
     return TypeUtil.visit(schema, new BuildPositionAccessors());
   }

From 00d5f9c9578a1a7825aeb6209c08dc108fe0b0a3 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Mon, 18 May 2020 11:44:42 +0800
Subject: [PATCH 07/10] refine code

---
 .../org/apache/iceberg/expressions/Binder.java     | 14 +++++++-------
 .../org/apache/iceberg/expressions/Evaluator.java  |  4 ++--
 .../org/apache/iceberg/data/GenericEvaluator.java  |  6 ++++--
 .../org/apache/iceberg/data/TableScanIterable.java |  5 +----
 4 files changed, 14 insertions(+), 15 deletions(-)

diff --git a/api/src/main/java/org/apache/iceberg/expressions/Binder.java b/api/src/main/java/org/apache/iceberg/expressions/Binder.java
index 23196a2a6baf..af6ee1415ce0 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Binder.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Binder.java
@@ -80,7 +80,7 @@ public static Expression bind(StructType struct,
    *
    * @param struct The {@link StructType struct type} to resolve references by name.
    * @param expr An {@link Expression expression} to rewrite with bound references.
-   * @param accessors A map to {@link Accessor accessor} to access the {@link StructLike StructLike} value
+   * @param idToAccessor A map that map the type ID to the value {@link Accessor accessor}
    * @param caseSensitive A boolean flag to control whether the bind should enforce case sensitivity.
    * @return the expression rewritten with bound references
    * @throws ValidationException if literals do not match bound references
@@ -88,9 +88,9 @@ public static Expression bind(StructType struct,
    */
   public static Expression bind(StructType struct,
                                 Expression expr,
-                                Map<Integer, Accessor<StructLike>> accessors,
+                                Map<Integer, Accessor<StructLike>> idToAccessor,
                                 boolean caseSensitive) {
-    return ExpressionVisitors.visit(expr, new BindVisitor(struct, accessors, caseSensitive));
+    return ExpressionVisitors.visit(expr, new BindVisitor(struct, idToAccessor, caseSensitive));
   }
 
   /**
@@ -135,17 +135,17 @@ public static Set<Integer> boundReferences(StructType struct, List<Expression> e
   private static class BindVisitor extends ExpressionVisitor<Expression> {
     private final StructType struct;
     private final boolean caseSensitive;
-    private Map<Integer, Accessor<StructLike>> accessors = null;
+    private Map<Integer, Accessor<StructLike>> idToAccessor = null;
 
     private BindVisitor(StructType struct, boolean caseSensitive) {
       this.struct = struct;
       this.caseSensitive = caseSensitive;
     }
 
-    private BindVisitor(StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive) {
+    private BindVisitor(StructType struct, Map<Integer, Accessor<StructLike>> idToAccessor, boolean caseSensitive) {
       this.struct = struct;
       this.caseSensitive = caseSensitive;
-      this.accessors = accessors;
+      this.idToAccessor = idToAccessor;
     }
 
     @Override
@@ -180,7 +180,7 @@ public <T> Expression predicate(BoundPredicate<T> pred) {
 
     @Override
     public <T> Expression predicate(UnboundPredicate<T> pred) {
-      return accessors == null ? pred.bind(struct, caseSensitive) : pred.bind(struct, accessors, caseSensitive);
+      return idToAccessor == null ? pred.bind(struct, caseSensitive) : pred.bind(struct, idToAccessor, caseSensitive);
     }
   }
 
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
index 878b73d0703c..0f48daf3d7e9 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
@@ -54,9 +54,9 @@ public Evaluator(StructType struct, Expression unbound, boolean caseSensitive) {
     this.expr = Binder.bind(struct, unbound, caseSensitive);
   }
 
-  public Evaluator(StructType struct, Expression unbound, Map<Integer, Accessor<StructLike>> accessors,
+  public Evaluator(StructType struct, Expression unbound, Map<Integer, Accessor<StructLike>> idToAccessor,
                    boolean caseSensitive) {
-    this.expr = Binder.bind(struct, unbound, accessors, caseSensitive);
+    this.expr = Binder.bind(struct, unbound, idToAccessor, caseSensitive);
   }
 
   public boolean eval(StructLike data) {
diff --git a/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java b/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
index 88c078597ba7..c8ebffdb0ad2 100644
--- a/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
+++ b/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
@@ -27,8 +27,10 @@
 import org.apache.iceberg.types.Types;
 
 public class GenericEvaluator extends Evaluator {
-  public GenericEvaluator(Types.StructType structType, Expression unbound, Map<Integer, Accessor<StructLike>> accessors,
+  public GenericEvaluator(Types.StructType structType,
+                          Expression unbound,
+                          Map<Integer, Accessor<StructLike>> idToAccessor,
                           boolean caseSensitive) {
-    super(structType, unbound, accessors, caseSensitive);
+    super(structType, unbound, idToAccessor, caseSensitive);
   }
 }
diff --git a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
index 9056aa3c9683..0ba300f941c6 100644
--- a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
+++ b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
@@ -29,12 +29,10 @@
 import java.util.Map;
 import java.util.NoSuchElementException;
 import org.apache.avro.generic.GenericData;
-import org.apache.iceberg.Accessor;
 import org.apache.iceberg.CombinedScanTask;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.avro.Avro;
@@ -162,9 +160,8 @@ public boolean hasNext() {
           this.currentCloseable = reader;
 
           if (task.residual() != null && task.residual() != Expressions.alwaysTrue()) {
-            Map<Integer, Accessor<StructLike>> accessors = Accessors.forSchema(projection);
             Evaluator filter = new GenericEvaluator(projection.asStruct(), task.residual(),
-                accessors, caseSensitive);
+                Accessors.forSchema(projection), caseSensitive);
             this.currentIterator = Iterables.filter(reader, filter::eval).iterator();
           } else {
             this.currentIterator = reader.iterator();

From 6865f139e28c854206ecd154b541ef548a60d5ec Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Sat, 23 May 2020 08:35:31 +0800
Subject: [PATCH 08/10] update to use wrapped class

---
 .../apache/iceberg/expressions/Binder.java    |  40 +--
 .../apache/iceberg/expressions/Evaluator.java |   7 -
 .../iceberg/expressions/NamedReference.java   |  15 +-
 .../apache/iceberg/expressions/Unbound.java   |  13 -
 .../iceberg/expressions/UnboundPredicate.java |  20 +-
 .../iceberg/expressions/UnboundTransform.java |  12 +-
 .../org/apache/iceberg/data/Accessors.java    | 278 ------------------
 .../apache/iceberg/data/GenericEvaluator.java |  36 ---
 .../iceberg/data/InternalRecordWrapper.java   |  88 ++++++
 .../iceberg/data/TableScanIterable.java       |   8 +-
 10 files changed, 97 insertions(+), 420 deletions(-)
 delete mode 100644 data/src/main/java/org/apache/iceberg/data/Accessors.java
 delete mode 100644 data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
 create mode 100644 data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java

diff --git a/api/src/main/java/org/apache/iceberg/expressions/Binder.java b/api/src/main/java/org/apache/iceberg/expressions/Binder.java
index af6ee1415ce0..1bc6563b0bc4 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Binder.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Binder.java
@@ -22,10 +22,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
-import org.apache.iceberg.Accessor;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.ExpressionVisitors.ExpressionVisitor;
 import org.apache.iceberg.types.Type;
@@ -65,34 +62,6 @@ public static Expression bind(StructType struct,
     return ExpressionVisitors.visit(expr, new BindVisitor(struct, caseSensitive));
   }
 
-  /**
-   * Replaces all unbound/named references with bound references to fields in the given struct.
-   * <p>
-   * When a reference is resolved, any literal used in a predicate for that field is converted to
-   * the field's type using {@link Literal#to(Type)}. If automatic conversion to that type isn't
-   * allowed, a {@link ValidationException validation exception} is thrown.
-   * <p>
-   * The result expression may be simplified when constructed. For example, {@code isNull("a")} is
-   * replaced with {@code alwaysFalse()} when {@code "a"} is resolved to a required field.
-   * <p>
-   * The expression cannot contain references that are already bound, or an
-   * {@link IllegalStateException} will be thrown.
-   *
-   * @param struct The {@link StructType struct type} to resolve references by name.
-   * @param expr An {@link Expression expression} to rewrite with bound references.
-   * @param idToAccessor A map that map the type ID to the value {@link Accessor accessor}
-   * @param caseSensitive A boolean flag to control whether the bind should enforce case sensitivity.
-   * @return the expression rewritten with bound references
-   * @throws ValidationException if literals do not match bound references
-   * @throws IllegalStateException if any references are already bound
-   */
-  public static Expression bind(StructType struct,
-                                Expression expr,
-                                Map<Integer, Accessor<StructLike>> idToAccessor,
-                                boolean caseSensitive) {
-    return ExpressionVisitors.visit(expr, new BindVisitor(struct, idToAccessor, caseSensitive));
-  }
-
   /**
    * Replaces all unbound/named references with bound references to fields in the given struct,
    * defaulting to case sensitive mode.
@@ -135,19 +104,12 @@ public static Set<Integer> boundReferences(StructType struct, List<Expression> e
   private static class BindVisitor extends ExpressionVisitor<Expression> {
     private final StructType struct;
     private final boolean caseSensitive;
-    private Map<Integer, Accessor<StructLike>> idToAccessor = null;
 
     private BindVisitor(StructType struct, boolean caseSensitive) {
       this.struct = struct;
       this.caseSensitive = caseSensitive;
     }
 
-    private BindVisitor(StructType struct, Map<Integer, Accessor<StructLike>> idToAccessor, boolean caseSensitive) {
-      this.struct = struct;
-      this.caseSensitive = caseSensitive;
-      this.idToAccessor = idToAccessor;
-    }
-
     @Override
     public Expression alwaysTrue() {
       return Expressions.alwaysTrue();
@@ -180,7 +142,7 @@ public <T> Expression predicate(BoundPredicate<T> pred) {
 
     @Override
     public <T> Expression predicate(UnboundPredicate<T> pred) {
-      return idToAccessor == null ? pred.bind(struct, caseSensitive) : pred.bind(struct, idToAccessor, caseSensitive);
+      return pred.bind(struct, caseSensitive);
     }
   }
 
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
index 0f48daf3d7e9..0d3e1ef311f8 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Evaluator.java
@@ -21,9 +21,7 @@
 
 import java.io.Serializable;
 import java.util.Comparator;
-import java.util.Map;
 import java.util.Set;
-import org.apache.iceberg.Accessor;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.expressions.ExpressionVisitors.BoundVisitor;
 import org.apache.iceberg.types.Types.StructType;
@@ -54,11 +52,6 @@ public Evaluator(StructType struct, Expression unbound, boolean caseSensitive) {
     this.expr = Binder.bind(struct, unbound, caseSensitive);
   }
 
-  public Evaluator(StructType struct, Expression unbound, Map<Integer, Accessor<StructLike>> idToAccessor,
-                   boolean caseSensitive) {
-    this.expr = Binder.bind(struct, unbound, idToAccessor, caseSensitive);
-  }
-
   public boolean eval(StructLike data) {
     return visitor().eval(data);
   }
diff --git a/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java b/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
index 552f1dbef34f..d928b3abfe1d 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/NamedReference.java
@@ -20,10 +20,7 @@
 package org.apache.iceberg.expressions;
 
 import com.google.common.base.Preconditions;
-import java.util.Map;
-import org.apache.iceberg.Accessor;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.types.Types;
 
@@ -41,12 +38,6 @@ public String name() {
 
   @Override
   public BoundReference<T> bind(Types.StructType struct, boolean caseSensitive) {
-    return bind(struct, null, caseSensitive);
-  }
-
-  @Override
-  public BoundReference<T> bind(Types.StructType struct, Map<Integer, Accessor<StructLike>> accessors,
-                                boolean caseSensitive) {
     Schema schema = new Schema(struct.fields());
     Types.NestedField field = caseSensitive ?
         schema.findField(name) :
@@ -55,11 +46,7 @@ public BoundReference<T> bind(Types.StructType struct, Map<Integer, Accessor<Str
     ValidationException.check(field != null,
         "Cannot find field '%s' in struct: %s", name, schema.asStruct());
 
-    if (accessors != null) {
-      return new BoundReference<>(field, accessors.get(field.fieldId()));
-    } else {
-      return new BoundReference<>(field, schema.accessorForField(field.fieldId()));
-    }
+    return new BoundReference<>(field, schema.accessorForField(field.fieldId()));
   }
 
   @Override
diff --git a/api/src/main/java/org/apache/iceberg/expressions/Unbound.java b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
index ac50e3b1949b..f160f935fb3e 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/Unbound.java
@@ -19,9 +19,6 @@
 
 package org.apache.iceberg.expressions;
 
-import java.util.Map;
-import org.apache.iceberg.Accessor;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.types.Types;
 
 /**
@@ -39,16 +36,6 @@ public interface Unbound<T, B> {
    */
   B bind(Types.StructType struct, boolean caseSensitive);
 
-  /**
-   * Bind this value expression to concrete types.
-   *
-   * @param struct input data types
-   * @param accessors the accessors for the struct like value
-   * @param caseSensitive whether binding should match columns using case sensitive resolution
-   * @return a bound value expression
-   */
-  B bind(Types.StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive);
-
   /**
    * @return this expression's underlying reference
    */
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
index 5a0400c514a2..fbef5e158934 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundPredicate.java
@@ -25,10 +25,7 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
-import org.apache.iceberg.Accessor;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.CharSequenceSet;
@@ -105,22 +102,7 @@ Expression bind(StructType struct) {
    */
   @Override
   public Expression bind(StructType struct, boolean caseSensitive) {
-    return bind(struct, null, caseSensitive);
-  }
-
-  /**
-   * Bind this UnboundPredicate.
-   *
-   * @param struct The {@link StructType struct type} to resolve references by name.
-   * @param accessors The accessors used to access values in struct
-   * @param caseSensitive A boolean flag to control whether the bind should enforce case sensitivity.
-   * @return an {@link Expression}
-   * @throws ValidationException if literals do not match bound references, or if comparison on expression is invalid
-   */
-  @Override
-  public Expression bind(StructType struct, Map<Integer, Accessor<StructLike>> accessors, boolean caseSensitive) {
-    BoundTerm<T> bound =
-        accessors == null ? term().bind(struct, caseSensitive) : term().bind(struct, accessors, caseSensitive);
+    BoundTerm<T> bound = term().bind(struct, caseSensitive);
 
     if (literals == null) {
       return bindUnaryOperation(bound);
diff --git a/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java b/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
index ea0ff9205f04..be3cec188935 100644
--- a/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
+++ b/api/src/main/java/org/apache/iceberg/expressions/UnboundTransform.java
@@ -19,9 +19,6 @@
 
 package org.apache.iceberg.expressions;
 
-import java.util.Map;
-import org.apache.iceberg.Accessor;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
@@ -48,14 +45,7 @@ public Transform<S, T> transform() {
   @SuppressWarnings("unchecked")
   @Override
   public BoundTransform<S, T> bind(Types.StructType struct, boolean caseSensitive) {
-    return bind(struct, null, caseSensitive);
-  }
-
-  @Override
-  public BoundTransform<S, T> bind(Types.StructType struct, Map<Integer, Accessor<StructLike>> accessors,
-                                   boolean caseSensitive) {
-    BoundReference<S> boundRef =
-        accessors == null ? ref.bind(struct, caseSensitive) : ref.bind(struct, accessors, caseSensitive);
+    BoundReference<S> boundRef = ref.bind(struct, caseSensitive);
 
     Transform<S, T> typeTransform;
     try {
diff --git a/data/src/main/java/org/apache/iceberg/data/Accessors.java b/data/src/main/java/org/apache/iceberg/data/Accessors.java
deleted file mode 100644
index 4172c4d14da2..000000000000
--- a/data/src/main/java/org/apache/iceberg/data/Accessors.java
+++ /dev/null
@@ -1,278 +0,0 @@
-/*
- * 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.data;
-
-import com.google.common.collect.Maps;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.OffsetDateTime;
-import java.util.List;
-import java.util.Map;
-import org.apache.iceberg.Accessor;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.StructLike;
-import org.apache.iceberg.types.Type;
-import org.apache.iceberg.types.TypeUtil;
-import org.apache.iceberg.types.Types;
-
-public class Accessors {
-  private Accessors() {
-  }
-
-  static Map<Integer, Accessor<StructLike>> forSchema(Schema schema) {
-    return TypeUtil.visit(schema, new BuildPositionAccessors());
-  }
-
-  private static class PositionAccessor implements Accessor<StructLike> {
-    private int position;
-    private final Type type;
-    private final Class<?> javaClass;
-
-    PositionAccessor(int pos, Type type) {
-      this.position = pos;
-      this.type = type;
-      this.javaClass = type.typeId().javaClass();
-    }
-
-    @Override
-    public Object get(StructLike row) {
-      return row.get(position, javaClass);
-    }
-
-    @Override
-    public Type type() {
-      return type;
-    }
-
-    public int position() {
-      return position;
-    }
-
-    public Class<?> javaClass() {
-      return javaClass;
-    }
-
-    @Override
-    public String toString() {
-      return "Accessor(positions=[" + position + "], type=" + type + ")";
-    }
-  }
-
-  private static class Position2Accessor implements Accessor<StructLike> {
-    private final int p0;
-    private final int p1;
-    private final Type type;
-    private final Class<?> javaClass;
-
-    Position2Accessor(int pos, PositionAccessor wrapped) {
-      this.p0 = pos;
-      this.p1 = wrapped.position();
-      this.type = wrapped.type();
-      this.javaClass = wrapped.javaClass();
-    }
-
-    @Override
-    public Object get(StructLike row) {
-      return row.get(p0, StructLike.class).get(p1, javaClass);
-    }
-
-    @Override
-    public Type type() {
-      return type;
-    }
-
-    public Class<?> javaClass() {
-      return javaClass;
-    }
-
-    @Override
-    public String toString() {
-      return "Accessor(positions=[" + p0 + ", " + p1 + "], type=" + type + ")";
-    }
-  }
-
-  private static class Position3Accessor implements Accessor<StructLike> {
-    private final int p0;
-    private final int p1;
-    private final int p2;
-    private final Type type;
-    private final Class<?> javaClass;
-
-    Position3Accessor(int pos, Position2Accessor wrapped) {
-      this.p0 = pos;
-      this.p1 = wrapped.p0;
-      this.p2 = wrapped.p1;
-      this.type = wrapped.type();
-      this.javaClass = wrapped.javaClass();
-    }
-
-    @Override
-    public Object get(StructLike row) {
-      return row.get(p0, StructLike.class).get(p1, StructLike.class).get(p2, javaClass);
-    }
-
-    @Override
-    public Type type() {
-      return type;
-    }
-
-    @Override
-    public String toString() {
-      return "Accessor(positions=[" + p0 + ", " + p1 + ", " + p2 + "], type=" + type + ")";
-    }
-  }
-
-  private static class WrappedPositionAccessor implements Accessor<StructLike> {
-    private final int position;
-    private final Accessor<StructLike> accessor;
-
-    WrappedPositionAccessor(int pos, Accessor<StructLike> accessor) {
-      this.position = pos;
-      this.accessor = accessor;
-    }
-
-    @Override
-    public Object get(StructLike row) {
-      StructLike inner = row.get(position, StructLike.class);
-      if (inner != null) {
-        return accessor.get(inner);
-      }
-      return null;
-    }
-
-    @Override
-    public Type type() {
-      return accessor.type();
-    }
-
-    @Override
-    public String toString() {
-      return "WrappedAccessor(position=" + position + ", wrapped=" + accessor + ")";
-    }
-  }
-
-  private static Accessor<StructLike> newAccessor(int pos, Type type) {
-    switch (type.typeId()) {
-      case DATE:
-        return new DateAccessor(pos, type);
-      case TIME:
-        return new TimeAccessor(pos, type);
-      case TIMESTAMP:
-        return new TimeStampAccessor(pos, type);
-      default:
-        return new PositionAccessor(pos, type);
-    }
-  }
-
-  private static Accessor<StructLike> newAccessor(int pos, boolean isOptional,
-                                                  Accessor<StructLike> accessor) {
-    if (isOptional) {
-      // the wrapped position handles null layers
-      return new WrappedPositionAccessor(pos, accessor);
-    } else if (accessor instanceof PositionAccessor) {
-      return new Position2Accessor(pos, (PositionAccessor) accessor);
-    } else if (accessor instanceof Position2Accessor) {
-      return new Position3Accessor(pos, (Position2Accessor) accessor);
-    } else {
-      return new WrappedPositionAccessor(pos, accessor);
-    }
-  }
-
-  private static class DateAccessor extends PositionAccessor {
-    private DateAccessor(int pos, Type type) {
-      super(pos, type);
-    }
-
-    @Override
-    public Object get(StructLike record) {
-      LocalDate value = record.get(position(), LocalDate.class);
-      return DateTimeUtil.daysFromDate(value);
-    }
-  }
-
-  private static class TimeAccessor extends PositionAccessor {
-    private TimeAccessor(int pos, Type type) {
-      super(pos, type);
-    }
-
-    @Override
-    public Object get(StructLike record) {
-      LocalTime value = record.get(position(), LocalTime.class);
-      return DateTimeUtil.microsFromTime(value);
-    }
-  }
-
-  private static class TimeStampAccessor extends PositionAccessor {
-    private TimeStampAccessor(int pos, Type type) {
-      super(pos, type);
-    }
-
-    @Override
-    public Object get(StructLike record) {
-      if (((Types.TimestampType) type().asPrimitiveType()).shouldAdjustToUTC()) {
-        OffsetDateTime value = record.get(position(), OffsetDateTime.class);
-        return DateTimeUtil.microsFromTimestamptz(value);
-      } else {
-        LocalDateTime value = record.get(position(), LocalDateTime.class);
-        return DateTimeUtil.microsFromTimestamp(value);
-      }
-    }
-  }
-
-  private static class BuildPositionAccessors extends TypeUtil.SchemaVisitor<Map<Integer, Accessor<StructLike>>> {
-
-    @Override
-    public Map<Integer, Accessor<StructLike>> schema(
-        Schema schema, Map<Integer, Accessor<StructLike>> structResult) {
-      return structResult;
-    }
-
-    @Override
-    public Map<Integer, Accessor<StructLike>> struct(
-        Types.StructType struct, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
-      Map<Integer, Accessor<StructLike>> accessors = Maps.newHashMap();
-      List<Types.NestedField> fields = struct.fields();
-      for (int i = 0; i < fieldResults.size(); i += 1) {
-        Types.NestedField field = fields.get(i);
-        Map<Integer, Accessor<StructLike>> result = fieldResults.get(i);
-        if (result != null) {
-          for (Map.Entry<Integer, Accessor<StructLike>> entry : result.entrySet()) {
-            accessors.put(entry.getKey(), newAccessor(i, field.isOptional(), entry.getValue()));
-          }
-        } else {
-          accessors.put(field.fieldId(), newAccessor(i, field.type()));
-        }
-      }
-
-      if (accessors.isEmpty()) {
-        return null;
-      }
-
-      return accessors;
-    }
-
-    @Override
-    public Map<Integer, Accessor<StructLike>> field(
-        Types.NestedField field, Map<Integer, Accessor<StructLike>> fieldResult) {
-      return fieldResult;
-    }
-  }
-}
diff --git a/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java b/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
deleted file mode 100644
index c8ebffdb0ad2..000000000000
--- a/data/src/main/java/org/apache/iceberg/data/GenericEvaluator.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.data;
-
-import java.util.Map;
-import org.apache.iceberg.Accessor;
-import org.apache.iceberg.StructLike;
-import org.apache.iceberg.expressions.Evaluator;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.types.Types;
-
-public class GenericEvaluator extends Evaluator {
-  public GenericEvaluator(Types.StructType structType,
-                          Expression unbound,
-                          Map<Integer, Accessor<StructLike>> idToAccessor,
-                          boolean caseSensitive) {
-    super(structType, unbound, idToAccessor, caseSensitive);
-  }
-}
diff --git a/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java b/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java
new file mode 100644
index 000000000000..c9efb3d41e67
--- /dev/null
+++ b/data/src/main/java/org/apache/iceberg/data/InternalRecordWrapper.java
@@ -0,0 +1,88 @@
+/*
+ * 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.data;
+
+import java.lang.reflect.Array;
+import java.nio.ByteBuffer;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.util.function.Function;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+class InternalRecordWrapper implements StructLike {
+  private final Function<Object, Object>[] transforms;
+  private StructLike wrapped = null;
+
+  @SuppressWarnings("unchecked")
+  InternalRecordWrapper(Types.StructType struct) {
+    this.transforms = struct.fields().stream()
+        .map(field -> converter(field.type()))
+        .toArray(length -> (Function<Object, Object>[]) Array.newInstance(Function.class, length));
+  }
+
+  private static Function<Object, Object> converter(Type type) {
+    switch (type.typeId()) {
+      case DATE:
+        return date -> DateTimeUtil.daysFromDate((LocalDate) date);
+      case TIME:
+        return time -> DateTimeUtil.microsFromTime((LocalTime) time);
+      case TIMESTAMP:
+        if (((Types.TimestampType) type).shouldAdjustToUTC()) {
+          return timestamp -> DateTimeUtil.microsFromTimestamptz((OffsetDateTime) timestamp);
+        } else {
+          return timestamp -> DateTimeUtil.microsFromTimestamp((LocalDateTime) timestamp);
+        }
+      case FIXED:
+        return bytes -> ByteBuffer.wrap((byte[]) bytes);
+      case STRUCT:
+        InternalRecordWrapper wrapper = new InternalRecordWrapper(type.asStructType());
+        return struct -> wrapper.wrap((StructLike) struct);
+      default:
+    }
+    return null;
+  }
+
+  public InternalRecordWrapper wrap(StructLike record) {
+    this.wrapped = record;
+    return this;
+  }
+
+  @Override
+  public int size() {
+    return wrapped.size();
+  }
+
+  @Override
+  public <T> T get(int pos, Class<T> javaClass) {
+    if (transforms[pos] != null) {
+      return javaClass.cast(transforms[pos].apply(wrapped.get(pos, Object.class)));
+    }
+    return wrapped.get(pos, javaClass);
+  }
+
+  @Override
+  public <T> void set(int pos, T value) {
+    throw new UnsupportedOperationException("Cannot update InternalRecordWrapper");
+  }
+}
diff --git a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
index 0ba300f941c6..51af01c040b7 100644
--- a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
+++ b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
@@ -160,9 +160,11 @@ public boolean hasNext() {
           this.currentCloseable = reader;
 
           if (task.residual() != null && task.residual() != Expressions.alwaysTrue()) {
-            Evaluator filter = new GenericEvaluator(projection.asStruct(), task.residual(),
-                Accessors.forSchema(projection), caseSensitive);
-            this.currentIterator = Iterables.filter(reader, filter::eval).iterator();
+            Evaluator filter = new Evaluator(projection.asStruct(), task.residual(), caseSensitive);
+            this.currentIterator = Iterables.filter(reader, record -> {
+              InternalRecordWrapper wrapperRecord = new InternalRecordWrapper(record.struct()).wrap(record);
+              return filter.eval(wrapperRecord);
+            }).iterator();
           } else {
             this.currentIterator = reader.iterator();
           }

From 34326405a11602cffa277cbd55f4228285ef6353 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Tue, 26 May 2020 09:56:16 +0800
Subject: [PATCH 09/10] address comments

---
 .../java/org/apache/iceberg/data/TableScanIterable.java   | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
index 51af01c040b7..c957ab5cb3b8 100644
--- a/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
+++ b/data/src/main/java/org/apache/iceberg/data/TableScanIterable.java
@@ -133,11 +133,13 @@ private class ScanIterator implements Iterator<Record>, Closeable {
     private final boolean caseSensitive;
     private Closeable currentCloseable = null;
     private Iterator<Record> currentIterator = Collections.emptyIterator();
+    private final InternalRecordWrapper recordWrapper;
 
     private ScanIterator(CloseableIterable<CombinedScanTask> tasks, boolean caseSensitive) {
       this.tasks = Lists.newArrayList(Iterables.concat(
           CloseableIterable.transform(tasks, CombinedScanTask::files))).iterator();
       this.caseSensitive = caseSensitive;
+      this.recordWrapper = new InternalRecordWrapper(projection.asStruct());
     }
 
     @Override
@@ -161,10 +163,8 @@ public boolean hasNext() {
 
           if (task.residual() != null && task.residual() != Expressions.alwaysTrue()) {
             Evaluator filter = new Evaluator(projection.asStruct(), task.residual(), caseSensitive);
-            this.currentIterator = Iterables.filter(reader, record -> {
-              InternalRecordWrapper wrapperRecord = new InternalRecordWrapper(record.struct()).wrap(record);
-              return filter.eval(wrapperRecord);
-            }).iterator();
+            this.currentIterator = Iterables.filter(reader,
+                record -> filter.eval(recordWrapper.wrap(record))).iterator();
           } else {
             this.currentIterator = reader.iterator();
           }

From cc189d84f07bcca16b36cacf6249f973eaee9634 Mon Sep 17 00:00:00 2001
From: Junjie Chen <chenjunjiedada@gmail.com>
Date: Tue, 26 May 2020 14:26:42 +0800
Subject: [PATCH 10/10] Assume false for orc

---
 data/src/test/java/org/apache/iceberg/data/TestLocalScan.java | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
index c15f59fc202e..b13a3dffe113 100644
--- a/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
+++ b/data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
@@ -56,6 +56,7 @@
 import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.types.Types;
 import org.junit.Assert;
+import org.junit.Assume;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -456,6 +457,7 @@ private DataFile writeFile(String location, String filename, Schema schema, List
 
   @Test
   public void testFilterWithDateAndTimestamp() throws IOException {
+    Assume.assumeFalse(format == FileFormat.ORC);
     Schema schema = new Schema(
         required(1, "timestamp_with_zone", Types.TimestampType.withZone()),
         required(2, "timestamp_without_zone", Types.TimestampType.withoutZone()),