API, Core: Add default value APIs and Avro implementation

.palantir/revapi.yml

@@ -873,6 +873,11 @@ acceptedBreaks:
       new: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
       justification: "Static utility class - should not have public constructor"
   "1.4.0":
+    org.apache.iceberg:iceberg-api:
+    - code: "java.class.defaultSerializationChanged"
+      old: "class org.apache.iceberg.types.Types.NestedField"
+      new: "class org.apache.iceberg.types.Types.NestedField"
+      justification: "Add default value APIs."
     org.apache.iceberg:iceberg-core:
     - code: "java.class.defaultSerializationChanged"
       old: "class org.apache.iceberg.PartitionData"

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

@@ -309,6 +309,18 @@ public static <T> UnboundTerm<T> transform(String name, Transform<?, T> transfor
     return new UnboundTransform<>(ref(name), transform);
   }
 
+  /**
+   * Create a {@link Literal} from an Object.
+   *
+   * @param value a value
+   * @param <T> Java type of value
+   * @return a Literal for the given value
+   * @throws IllegalArgumentException if the value has no literal implementation
+   */
+  public static <T> Literal<T> lit(T value) {
+    return Literals.from(value);
+  }
+
   public static <T> UnboundAggregate<T> count(String name) {
     return new UnboundAggregate<>(Operation.COUNT, ref(name));
   }

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

@@ -27,6 +27,7 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -414,43 +415,132 @@ public int hashCode() {
 
   public static class NestedField implements Serializable {
     public static NestedField optional(int id, String name, Type type) {
-      return new NestedField(true, id, name, type, null);
+      return new NestedField(true, id, name, type, null, null, null);
     }
 
     public static NestedField optional(int id, String name, Type type, String doc) {
-      return new NestedField(true, id, name, type, doc);
+      return new NestedField(true, id, name, type, doc, null, null);
     }
 
     public static NestedField required(int id, String name, Type type) {
-      return new NestedField(false, id, name, type, null);
+      return new NestedField(false, id, name, type, null, null, null);
     }
 
     public static NestedField required(int id, String name, Type type, String doc) {
-      return new NestedField(false, id, name, type, doc);
+      return new NestedField(false, id, name, type, doc, null, null);
     }
 
     public static NestedField of(int id, boolean isOptional, String name, Type type) {
-      return new NestedField(isOptional, id, name, type, null);
+      return new NestedField(isOptional, id, name, type, null, null, null);
     }
 
     public static NestedField of(int id, boolean isOptional, String name, Type type, String doc) {
-      return new NestedField(isOptional, id, name, type, doc);
+      return new NestedField(isOptional, id, name, type, doc, null, null);
+    }
+
+    public static Builder from(NestedField field) {
+      return new Builder(field);
+    }
+
+    public static Builder required(String name) {
+      return new Builder(false, name);
+    }
+
+    public static Builder optional(String name) {
+      return new Builder(true, name);
+    }
+
+    public static class Builder {
+      private final boolean isOptional;
+      private final String name;
+      private Integer id = null;
+      private Type type = null;
+      private String doc = null;
+      private Object initialDefault = null;
+      private Object writeDefault = null;
+
+      private Builder(boolean isFieldOptional, String fieldName) {
+        isOptional = isFieldOptional;
+        name = fieldName;
+      }
+
+      private Builder(NestedField toCopy) {
+        this.isOptional = toCopy.isOptional;
+        this.name = toCopy.name;
+        this.id = toCopy.id;
+        this.type = toCopy.type;
+        this.doc = toCopy.doc;
+        this.initialDefault = toCopy.initialDefault;
+        this.writeDefault = toCopy.writeDefault;
+      }
+
+      public Builder withId(int fieldId) {
+        id = fieldId;
+        return this;
+      }
+
+      public Builder ofType(Type fieldType) {
+        type = fieldType;
+        return this;
+      }
+
+      public Builder withDoc(String fieldDoc) {
+        doc = fieldDoc;
+        return this;
+      }
+
+      public Builder withInitialDefault(Object fieldInitialDefault) {
+        initialDefault = fieldInitialDefault;
+        return this;
+      }
+
+      public Builder withWriteDefault(Object fieldWriteDefault) {
+        writeDefault = fieldWriteDefault;
+        return this;
+      }
+
+      public NestedField build() {
+        // the constructor validates the fields
+        return new NestedField(isOptional, id, name, type, doc, initialDefault, writeDefault);
+      }
     }
 
     private final boolean isOptional;
     private final int id;
     private final String name;
     private final Type type;
     private final String doc;
-
-    private NestedField(boolean isOptional, int id, String name, Type type, String doc) {
+    private final Object initialDefault;
+    private final Object writeDefault;
+
+    private NestedField(
+        boolean isOptional,
+        int id,
+        String name,
+        Type type,
+        String doc,
+        Object initialDefault,
+        Object writeDefault) {
       Preconditions.checkNotNull(name, "Name cannot be null");
       Preconditions.checkNotNull(type, "Type cannot be null");
       this.isOptional = isOptional;
       this.id = id;
       this.name = name;
       this.type = type;
       this.doc = doc;
+      this.initialDefault = castDefault(initialDefault, type);
+      this.writeDefault = castDefault(writeDefault, type);
+    }
+
+    private static Object castDefault(Object defaultValue, Type type) {
+      if (type.isNestedType() && defaultValue != null) {
+        throw new IllegalArgumentException(
+            String.format("Invalid default value for %s: %s (must be null)", type, defaultValue));
+      } else if (defaultValue != null) {
+        return Expressions.lit(defaultValue).to(type).value();
+      }
+
+      return null;
     }
 
     public boolean isOptional() {
@@ -461,7 +551,7 @@ public NestedField asOptional() {
       if (isOptional) {
         return this;
       }
-      return new NestedField(true, id, name, type, doc);
+      return new NestedField(true, id, name, type, doc, initialDefault, writeDefault);
     }
 
     public boolean isRequired() {
@@ -472,11 +562,15 @@ public NestedField asRequired() {
       if (!isOptional) {
         return this;
       }
-      return new NestedField(false, id, name, type, doc);
+      return new NestedField(false, id, name, type, doc, initialDefault, writeDefault);
     }
 
+    /**
+     * @deprecated will be removed in 2.0.0; use {@link Builder#withId(int)} instead
+     */
+    @Deprecated
     public NestedField withFieldId(int newId) {
-      return new NestedField(isOptional, newId, name, type, doc);
+      return new NestedField(isOptional, newId, name, type, doc, initialDefault, writeDefault);
     }
 
     public int fieldId() {
@@ -495,6 +589,14 @@ public String doc() {
       return doc;
     }
 
+    public Object initialDefault() {
+      return initialDefault;
+    }
+
+    public Object writeDefault() {
+      return writeDefault;
+    }
+
     @Override
     public String toString() {
       return String.format("%d: %s: %s %s", id, name, isOptional ? "optional" : "required", type)

core/src/main/java/org/apache/iceberg/avro/GenericAvroReader.java

@@ -47,16 +47,16 @@ public class GenericAvroReader<T>
   private Schema fileSchema = null;
   private ValueReader<T> reader = null;
 
-  public static <D> GenericAvroReader<D> create(org.apache.iceberg.Schema schema) {
-    return new GenericAvroReader<>(schema);
+  public static <D> GenericAvroReader<D> create(org.apache.iceberg.Schema expectedSchema) {
+    return new GenericAvroReader<>(expectedSchema);
   }
 
-  public static <D> GenericAvroReader<D> create(Schema schema) {
-    return new GenericAvroReader<>(schema);
+  public static <D> GenericAvroReader<D> create(Schema readSchema) {
+    return new GenericAvroReader<>(readSchema);
   }
 
-  GenericAvroReader(org.apache.iceberg.Schema readSchema) {
-    this.expectedType = readSchema.asStruct();
+  GenericAvroReader(org.apache.iceberg.Schema expectedSchema) {
+    this.expectedType = expectedSchema.asStruct();
   }
 
   GenericAvroReader(Schema readSchema) {
@@ -140,6 +140,8 @@ public ValueReader<?> record(Type partner, Schema record, List<ValueReader<?>> f
         Types.NestedField field = expected.field(fieldId);
         if (constant != null) {
           readPlan.add(Pair.of(pos, ValueReaders.constant(constant)));
+        } else if (field.initialDefault() != null) {
+          readPlan.add(Pair.of(pos, ValueReaders.constant(field.initialDefault())));
         } else if (fieldId == MetadataColumns.IS_DELETED.fieldId()) {
           readPlan.add(Pair.of(pos, ValueReaders.constant(false)));
         } else if (fieldId == MetadataColumns.ROW_POSITION.fieldId()) {