PARQUET-34: Use Int for Size predicate value

Author: clairemcginty

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/FilterApi.java

@@ -264,7 +264,7 @@ public static <T extends Comparable<T>, P extends SingleColumnFilterPredicate<T>
     return Contains.of(pred);
   }
 
-  public static Size size(Column<?> column, Size.Operator operator, long value) {
+  public static Size size(Column<?> column, Size.Operator operator, int value) {
     return new Size(column, operator, value);
   }
 

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/LogicalInverter.java

@@ -101,7 +101,7 @@ public <T extends Comparable<T>> FilterPredicate visit(Contains<T> contains) {
 
   @Override
   public FilterPredicate visit(Size size) {
-    final long value = size.getValue();
+    final int value = size.getValue();
     final Operators.Column<?> column = size.getColumn();
 
     return size.filter(

parquet-column/src/main/java/org/apache/parquet/filter2/predicate/Operators.java

@@ -516,9 +516,9 @@ public enum Operator {
 
     private final Column<?> column;
     private final Operator operator;
-    private final long value;
+    private final int value;
 
-    Size(Column<?> column, Operator operator, long value) {
+    Size(Column<?> column, Operator operator, int value) {
       this.column = column;
       this.operator = operator;
       if (value < 0) {
@@ -532,7 +532,7 @@ public <R> R accept(Visitor<R> visitor) {
       return visitor.visit(this);
     }
 
-    public long getValue() {
+    public int getValue() {
       return value;
     }
 
@@ -541,11 +541,11 @@ public Column<?> getColumn() {
     }
 
     public <R> R filter(
-        Function<Long, R> onEq,
-        Function<Long, R> onLt,
-        Function<Long, R> onLtEq,
-        Function<Long, R> onGt,
-        Function<Long, R> onGtEq) {
+        Function<Integer, R> onEq,
+        Function<Integer, R> onLt,
+        Function<Integer, R> onLtEq,
+        Function<Integer, R> onGt,
+        Function<Integer, R> onGtEq) {
       if (operator == Operator.EQ) {
         return onEq.apply(value);
       } else if (operator == Operator.LT) {

parquet-column/src/main/java/org/apache/parquet/filter2/recordlevel/IncrementallyUpdatedFilterPredicate.java

@@ -225,7 +225,7 @@ public void reset() {
   }
 
   class CountingValueInspector extends ValueInspector {
-    private long observedValueCount;
+    private int observedValueCount;
     private final ValueInspector delegate;
 
     /**
@@ -236,9 +236,9 @@ class CountingValueInspector extends ValueInspector {
      * underlying `lt(3)` predicate to be evaluated on the first or second elements of the array, since it would
      * return a premature True value.
      */
-    private final Function<Long, Boolean> shouldUpdateDelegate;
+    private final Function<Integer, Boolean> shouldUpdateDelegate;
 
-    public CountingValueInspector(ValueInspector delegate, Function<Long, Boolean> shouldUpdateDelegate) {
+    public CountingValueInspector(ValueInspector delegate, Function<Integer, Boolean> shouldUpdateDelegate) {
       this.observedValueCount = 0;
       this.delegate = delegate;
       this.shouldUpdateDelegate = shouldUpdateDelegate;

parquet-column/src/main/java/org/apache/parquet/filter2/recordlevel/IncrementallyUpdatedFilterPredicateBuilderBase.java

@@ -60,7 +60,7 @@
  */
 public abstract class IncrementallyUpdatedFilterPredicateBuilderBase
     implements Visitor<IncrementallyUpdatedFilterPredicate> {
-  static final Operators.LongColumn SIZE_PSUEDOCOLUMN = FilterApi.longColumn("$SIZE");
+  static final Operators.IntColumn SIZE_PSUEDOCOLUMN = FilterApi.intColumn("$SIZE");
 
   private boolean built = false;
   private final Map<ColumnPath, List<ValueInspector>> valueInspectorsByColumn = new HashMap<>();
@@ -80,7 +80,7 @@ public IncrementallyUpdatedFilterPredicateBuilderBase(List<PrimitiveColumnIO> le
         SIZE_PSUEDOCOLUMN.getColumnPath(),
         new PrimitiveType(
                 Type.Repetition.REQUIRED,
-                PrimitiveType.PrimitiveTypeName.INT64,
+                PrimitiveType.PrimitiveTypeName.INT32,
                 SIZE_PSUEDOCOLUMN.getColumnPath().toDotString())
             .comparator());
   }