Add support for struct field based filtering

api/src/main/java/org/apache/iceberg/Accessor.java

@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+import org.apache.iceberg.types.Type;
+
+public interface Accessor<T> extends Serializable {
+  Object get(T container);
+
+  Type type();
+}

api/src/main/java/org/apache/iceberg/Accessors.java

@@ -0,0 +1,228 @@
+/*
+ * 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;
+
+import com.google.common.collect.Maps;
+import java.util.List;
+import java.util.Map;
+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) {
+    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 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;
+    }
+  }
+}

api/src/main/java/org/apache/iceberg/Schema.java

@@ -48,6 +48,7 @@ public class Schema implements Serializable {
   private transient Map<Integer, Types.NestedField> idToField = null;
   private transient BiMap<String, Integer> nameToId = null;
   private transient BiMap<String, Integer> lowerCaseNameToId = null;
+  private transient Map<Integer, Accessor<StructLike>> idToAccessor = null;
 
   public Schema(List<Types.NestedField> columns, Map<String, Integer> aliases) {
     this.struct = Types.StructType.of(columns);
@@ -58,6 +59,10 @@ public Schema(List<Types.NestedField> columns) {
     this.struct = Types.StructType.of(columns);
   }
 
+  public Schema(Types.NestedField... columns) {
+    this(Arrays.asList(columns));
+  }
+
   private Map<Integer, Types.NestedField> lazyIdToField() {
     if (idToField == null) {
       this.idToField = TypeUtil.indexById(struct);
@@ -79,8 +84,11 @@ private BiMap<String, Integer> lazyLowerCaseNameToId() {
     return lowerCaseNameToId;
   }
 
-  public Schema(Types.NestedField... columns) {
-    this(Arrays.asList(columns));
+  private Map<Integer, Accessor<StructLike>> lazyIdToAccessor() {
+    if (idToAccessor == null) {
+      idToAccessor = Accessors.forSchema(this);
+    }
+    return idToAccessor;
   }
 
   /**
@@ -141,9 +149,9 @@ public Types.NestedField findField(int id) {
   }
 
   /**
-   * Returns a sub-field field by name as a {@link Types.NestedField}.
+   * Returns a sub-field by name as a {@link Types.NestedField}.
    * <p>
-   * The result may be a nested field.
+   * The result may be a top-level or a nested field.
    *
    * @param name a String name
    * @return a Type for the sub-field or null if it is not found
@@ -157,6 +165,23 @@ public Types.NestedField findField(String name) {
     return null;
   }
 
+  /**
+   * Returns a sub-field by name as a {@link Types.NestedField}.
+   * <p>
+   * The result may be a top-level or a nested field.
+   *
+   * @param name a String name
+   * @return the sub-field or null if it is not found
+   */
+  public Types.NestedField caseInsensitiveFindField(String name) {
+    Preconditions.checkArgument(!name.isEmpty(), "Invalid column name: (empty)");
+    Integer id = lazyLowerCaseNameToId().get(name.toLowerCase(Locale.ROOT));
+    if (id != null) {
+      return lazyIdToField().get(id);
+    }
+    return null;
+  }
+
   /**
    * Returns the full column name for the given id.
    *
@@ -195,6 +220,18 @@ public String idToAlias(Integer fieldId) {
     return null;
   }
 
+  /**
+   * Return an accessor for retrieving the data from {@link StructLike}.
+   * <p>
+   * Accessors do not retrieve data contained in lists or maps.
+   *
+   * @param id a column id in this schema
+   * @return an {@link Accessor} to retrieve values from a {@link StructLike} row
+   */
+  public Accessor<StructLike> accessorForField(int id) {
+    return lazyIdToAccessor().get(id);
+  }
+
   /**
    * Creates a projection schema for a subset of columns, selected by name.
    * <p>