API, Core, Flink, Avro, Parquet: Remove dead code and update javadocs (…

…apache#10530)

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

@@ -484,7 +484,7 @@ public <T> String predicate(UnboundPredicate<T> pred) {
     }
   }
 
-  private static <T> List<String> abbreviateValues(List<String> sanitizedValues) {
+  private static List<String> abbreviateValues(List<String> sanitizedValues) {
     if (sanitizedValues.size() >= LONG_IN_PREDICATE_ABBREVIATION_THRESHOLD) {
       Set<String> distinctValues = ImmutableSet.copyOf(sanitizedValues);
       if (distinctValues.size()

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

@@ -190,10 +190,6 @@ public <T> Expression predicate(UnboundPredicate<T> pred) {
     PartitionSpec spec() {
       return spec;
     }
-
-    boolean isCaseSensitive() {
-      return caseSensitive;
-    }
   }
 
   private static class InclusiveProjection extends BaseProjectionEvaluator {

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/GenericArrowVectorAccessorFactory.java

@@ -561,12 +561,6 @@ private static class FixedSizeBinaryAccessor<
     private final FixedSizeBinaryVector vector;
     private final StringFactory<Utf8StringT> stringFactory;
 
-    FixedSizeBinaryAccessor(FixedSizeBinaryVector vector) {
-      super(vector);
-      this.vector = vector;
-      this.stringFactory = null;
-    }
-
     FixedSizeBinaryAccessor(
         FixedSizeBinaryVector vector, StringFactory<Utf8StringT> stringFactory) {
       super(vector);

core/src/main/java/org/apache/iceberg/DeleteFileIndex.java

@@ -737,22 +737,10 @@ public DeleteFile wrapped() {
       return wrapped;
     }
 
-    public PartitionSpec spec() {
-      return spec;
-    }
-
-    public StructLike partition() {
-      return wrapped.partition();
-    }
-
     public long applySequenceNumber() {
       return applySequenceNumber;
     }
 
-    public FileContent content() {
-      return wrapped.content();
-    }
-
     public List<Types.NestedField> equalityFields() {
       if (equalityFields == null) {
         synchronized (this) {
@@ -778,10 +766,6 @@ public Map<Integer, Long> nullValueCounts() {
       return wrapped.nullValueCounts();
     }
 
-    public Map<Integer, Long> nanValueCounts() {
-      return wrapped.nanValueCounts();
-    }
-
     public boolean hasLowerAndUpperBounds() {
       return wrapped.lowerBounds() != null && wrapped.upperBounds() != null;
     }

core/src/main/java/org/apache/iceberg/hadoop/HadoopStreams.java

@@ -21,7 +21,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.nio.ByteBuffer;
 import java.util.Arrays;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -123,10 +122,6 @@ public int read(byte[] b, int off, int len) throws IOException {
       return stream.read(b, off, len);
     }
 
-    public int read(ByteBuffer buf) throws IOException {
-      return stream.read(buf);
-    }
-
     @SuppressWarnings("checkstyle:NoFinalizer")
     @Override
     protected void finalize() throws Throwable {

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java

@@ -163,6 +163,8 @@ Map<SortKey, Long> mapStatistics() {
   }
 
   /**
+   * Returns assignment summary for every subtask.
+   *
    * @return assignment summary for every subtask. Key is subtaskId. Value pair is (weight assigned
    *     to the subtask, number of keys assigned to the subtask)
    */
@@ -328,7 +330,11 @@ static class KeyAssignment {
       }
     }
 
-    /** @return subtask id */
+    /**
+     * Select a subtask for the key.
+     *
+     * @return subtask id
+     */
     int select() {
       if (assignedSubtasks.length == 1) {
         // only choice. no need to run random number generator.

flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/EnumerationHistory.java

@@ -78,7 +78,11 @@ synchronized boolean hasFullHistory() {
     return count >= history.length;
   }
 
-  /** @return true if split discovery should pause because assigner has too many splits already. */
+  /**
+   * Checks whether split discovery should be paused.
+   *
+   * @return true if split discovery should pause because assigner has too many splits already.
+   */
   synchronized boolean shouldPauseSplitDiscovery(int pendingSplitCountFromAssigner) {
     if (count < history.length) {
       // only check throttling when full history is obtained.

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java

@@ -163,6 +163,8 @@ Map<SortKey, Long> mapStatistics() {
   }
 
   /**
+   * Returns assignment summary for every subtask.
+   *
    * @return assignment summary for every subtask. Key is subtaskId. Value pair is (weight assigned
    *     to the subtask, number of keys assigned to the subtask)
    */
@@ -328,7 +330,11 @@ static class KeyAssignment {
       }
     }
 
-    /** @return subtask id */
+    /**
+     * Select a subtask for the key.
+     *
+     * @return subtask id
+     */
     int select() {
       if (assignedSubtasks.length == 1) {
         // only choice. no need to run random number generator.

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/EnumerationHistory.java

@@ -78,7 +78,11 @@ synchronized boolean hasFullHistory() {
     return count >= history.length;
   }
 
-  /** @return true if split discovery should pause because assigner has too many splits already. */
+  /**
+   * Checks whether split discovery should be paused.
+   *
+   * @return true if split discovery should pause because assigner has too many splits already.
+   */
   synchronized boolean shouldPauseSplitDiscovery(int pendingSplitCountFromAssigner) {
     if (count < history.length) {
       // only check throttling when full history is obtained.

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/MapRangePartitioner.java

@@ -161,6 +161,8 @@ Map<SortKey, Long> mapStatistics() {
   }
 
   /**
+   * Returns assignment summary for every subtask.
+   *
    * @return assignment summary for every subtask. Key is subtaskId. Value pair is (weight assigned
    *     to the subtask, number of keys assigned to the subtask)
    */
@@ -326,7 +328,11 @@ static class KeyAssignment {
       }
     }
 
-    /** @return subtask id */
+    /**
+     * Select a subtask for the key.
+     *
+     * @return subtask id
+     */
     int select() {
       if (assignedSubtasks.length == 1) {
         // only choice. no need to run random number generator.

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/source/enumerator/EnumerationHistory.java

@@ -78,7 +78,11 @@ synchronized boolean hasFullHistory() {
     return count >= history.length;
   }
 
-  /** @return true if split discovery should pause because assigner has too many splits already. */
+  /**
+   * Checks whether split discovery should be paused.
+   *
+   * @return true if split discovery should pause because assigner has too many splits already.
+   */
   synchronized boolean shouldPauseSplitDiscovery(int pendingSplitCountFromAssigner) {
     if (count < history.length) {
       // only check throttling when full history is obtained.

parquet/src/main/java/org/apache/iceberg/parquet/ParquetAvro.java

@@ -377,10 +377,6 @@ private static class Pair<K, V> {
       this.second = second;
     }
 
-    public static <K, V> Pair<K, V> of(K first, V second) {
-      return new Pair<>(first, second);
-    }
-
     public K getFirst() {
       return first;
     }

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueReaders.java

@@ -696,28 +696,21 @@ public V setValue(V newValue) {
   }
 
   public abstract static class StructReader<T, I> implements ParquetValueReader<T> {
-    private interface Setter<R> {
-      void set(R record, int pos, Object reuse);
-    }
-
     private final ParquetValueReader<?>[] readers;
     private final TripleIterator<?> column;
     private final List<TripleIterator<?>> children;
 
-    @SuppressWarnings("unchecked")
     protected StructReader(List<Type> types, List<ParquetValueReader<?>> readers) {
       this.readers =
           (ParquetValueReader<?>[]) Array.newInstance(ParquetValueReader.class, readers.size());
       TripleIterator<?>[] columns =
           (TripleIterator<?>[]) Array.newInstance(TripleIterator.class, readers.size());
-      Setter<I>[] setters = (Setter<I>[]) Array.newInstance(Setter.class, readers.size());
 
       ImmutableList.Builder<TripleIterator<?>> columnsBuilder = ImmutableList.builder();
       for (int i = 0; i < readers.size(); i += 1) {
         ParquetValueReader<?> reader = readers.get(i);
         this.readers[i] = readers.get(i);
         columns[i] = reader.column();
-        setters[i] = newSetter(reader, types.get(i));
         columnsBuilder.addAll(reader.columns());
       }
 
@@ -754,41 +747,6 @@ public List<TripleIterator<?>> columns() {
       return children;
     }
 
-    @SuppressWarnings("unchecked")
-    private <E> Setter<I> newSetter(ParquetValueReader<E> reader, Type type) {
-      if (reader instanceof UnboxedReader && type.isPrimitive()) {
-        UnboxedReader<?> unboxed = (UnboxedReader<?>) reader;
-        switch (type.asPrimitiveType().getPrimitiveTypeName()) {
-          case BOOLEAN:
-            return (record, pos, ignored) -> setBoolean(record, pos, unboxed.readBoolean());
-          case INT32:
-            return (record, pos, ignored) -> setInteger(record, pos, unboxed.readInteger());
-          case INT64:
-            return (record, pos, ignored) -> setLong(record, pos, unboxed.readLong());
-          case FLOAT:
-            return (record, pos, ignored) -> setFloat(record, pos, unboxed.readFloat());
-          case DOUBLE:
-            return (record, pos, ignored) -> setDouble(record, pos, unboxed.readDouble());
-          case INT96:
-          case FIXED_LEN_BYTE_ARRAY:
-          case BINARY:
-            return (record, pos, ignored) -> set(record, pos, unboxed.readBinary());
-          default:
-            throw new UnsupportedOperationException("Unsupported type: " + type);
-        }
-      }
-
-      // TODO: Add support for options to avoid the null check
-      return (record, pos, reuse) -> {
-        Object obj = reader.read((E) reuse);
-        if (obj != null) {
-          set(record, pos, obj);
-        } else {
-          setNull(record, pos);
-        }
-      };
-    }
-
     @SuppressWarnings("unchecked")
     private <E> E get(I intermediate, int pos) {
       return (E) getField(intermediate, pos);

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java

@@ -147,18 +147,10 @@ private UnboxedWriter(ColumnDescriptor desc) {
       super(desc);
     }
 
-    public void writeBoolean(int repetitionLevel, boolean value) {
-      column.writeBoolean(repetitionLevel, value);
-    }
-
     public void writeInteger(int repetitionLevel, int value) {
       column.writeInteger(repetitionLevel, value);
     }
 
-    public void writeLong(int repetitionLevel, long value) {
-      column.writeLong(repetitionLevel, value);
-    }
-
     public void writeFloat(int repetitionLevel, float value) {
       column.writeFloat(repetitionLevel, value);
     }