[common] Optimize getNullCounts() to return int[] instead of Long[]

fluss-common/src/main/java/org/apache/fluss/predicate/And.java

@@ -51,7 +51,7 @@ public boolean test(
             long rowCount,
             InternalRow minValues,
             InternalRow maxValues,
-            Long[] nullCounts,
+            int[] nullCounts,
             List<Predicate> children) {
         for (Predicate child : children) {
             if (!child.test(rowCount, minValues, maxValues, nullCounts)) {

fluss-common/src/main/java/org/apache/fluss/predicate/CompoundPredicate.java

@@ -58,7 +58,7 @@ public boolean test(InternalRow row) {
 
     @Override
     public boolean test(
-            long rowCount, InternalRow minValues, InternalRow maxValues, Long[] nullCounts) {
+            long rowCount, InternalRow minValues, InternalRow maxValues, int[] nullCounts) {
         return function.test(rowCount, minValues, maxValues, nullCounts, children);
     }
 
@@ -102,7 +102,7 @@ public abstract boolean test(
                 long rowCount,
                 InternalRow minValues,
                 InternalRow maxValues,
-                Long[] nullCounts,
+                int[] nullCounts,
                 List<Predicate> children);
 
         public abstract Optional<Predicate> negate(List<Predicate> children);

fluss-common/src/main/java/org/apache/fluss/predicate/Contains.java

@@ -47,7 +47,7 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             Object patternLiteral) {
         return true;
     }

fluss-common/src/main/java/org/apache/fluss/predicate/EndsWith.java

@@ -51,7 +51,7 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             Object patternLiteral) {
         return true;
     }

fluss-common/src/main/java/org/apache/fluss/predicate/Equal.java

@@ -44,7 +44,7 @@ public boolean test(DataType type, Object field, Object literal) {
 
     @Override
     public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal) {
         return compareLiteral(type, literal, min) >= 0 && compareLiteral(type, literal, max) <= 0;
     }
 

fluss-common/src/main/java/org/apache/fluss/predicate/GreaterOrEqual.java

@@ -44,7 +44,7 @@ public boolean test(DataType type, Object field, Object literal) {
 
     @Override
     public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal) {
         return compareLiteral(type, literal, max) <= 0;
     }
 

fluss-common/src/main/java/org/apache/fluss/predicate/GreaterThan.java

@@ -44,7 +44,7 @@ public boolean test(DataType type, Object field, Object literal) {
 
     @Override
     public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal) {
         return compareLiteral(type, literal, max) < 0;
     }
 

fluss-common/src/main/java/org/apache/fluss/predicate/In.java

@@ -56,9 +56,9 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             List<Object> literals) {
-        if (nullCount != null && rowCount == nullCount) {
+        if (nullCount >= 0 && rowCount == nullCount) {
             return false;
         }
         for (Object literal : literals) {

fluss-common/src/main/java/org/apache/fluss/predicate/IsNotNull.java

@@ -41,8 +41,8 @@ public boolean test(DataType type, Object field) {
     }
 
     @Override
-    public boolean test(DataType type, long rowCount, Object min, Object max, Long nullCount) {
-        return nullCount == null || nullCount < rowCount;
+    public boolean test(DataType type, long rowCount, Object min, Object max, long nullCount) {
+        return nullCount < 0 || nullCount < rowCount;
     }
 
     @Override

fluss-common/src/main/java/org/apache/fluss/predicate/IsNull.java

@@ -41,8 +41,8 @@ public boolean test(DataType type, Object field) {
     }
 
     @Override
-    public boolean test(DataType type, long rowCount, Object min, Object max, Long nullCount) {
-        return nullCount == null || nullCount > 0;
+    public boolean test(DataType type, long rowCount, Object min, Object max, long nullCount) {
+        return nullCount < 0 || nullCount > 0;
     }
 
     @Override

fluss-common/src/main/java/org/apache/fluss/predicate/LeafFunction.java

@@ -49,7 +49,7 @@ public abstract class LeafFunction implements Serializable {
      * @param rowCount the total number of rows
      * @param min the minimum value of the field in the rows
      * @param max the maximum value of the field in the rows
-     * @param nullCount the number of null values in the field, or null if unknown
+     * @param nullCount the number of null values in the field, or -1 if unknown
      * @param literals the literals to test against the field
      * @return true if there is any row satisfies the condition, false otherwise
      */
@@ -58,7 +58,7 @@ public abstract boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             List<Object> literals);
 
     public abstract Optional<LeafFunction> negate();

fluss-common/src/main/java/org/apache/fluss/predicate/LeafPredicate.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.predicate;
 
+import org.apache.fluss.record.LogRecordBatchStatistics;
 import org.apache.fluss.row.InternalRow;
 import org.apache.fluss.types.DataType;
 import org.apache.fluss.types.DecimalType;
@@ -87,11 +88,14 @@ public boolean test(InternalRow row) {
 
     @Override
     public boolean test(
-            long rowCount, InternalRow minValues, InternalRow maxValues, Long[] nullCounts) {
+            long rowCount, InternalRow minValues, InternalRow maxValues, int[] nullCounts) {
         Object min = get(minValues, fieldIndex, type);
         Object max = get(maxValues, fieldIndex, type);
-        Long nullCount = nullCounts != null ? nullCounts[fieldIndex] : null;
-        if (nullCount == null || rowCount != nullCount) {
+        long nullCount =
+                nullCounts != null
+                        ? nullCounts[fieldIndex]
+                        : LogRecordBatchStatistics.NULL_COUNT_UNAVAILABLE;
+        if (nullCount < 0 || rowCount != nullCount) {
             // not all null
             // min or max is null
             // unknown stats

fluss-common/src/main/java/org/apache/fluss/predicate/LeafUnaryFunction.java

@@ -33,7 +33,7 @@ public abstract class LeafUnaryFunction extends LeafFunction {
     public abstract boolean test(DataType type, Object value);
 
     public abstract boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount);
+            DataType type, long rowCount, Object min, Object max, long nullCount);
 
     @Override
     public boolean test(DataType type, Object value, List<Object> literals) {
@@ -46,7 +46,7 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             List<Object> literals) {
         return test(type, rowCount, min, max, nullCount);
     }

fluss-common/src/main/java/org/apache/fluss/predicate/LessOrEqual.java

@@ -44,7 +44,7 @@ public boolean test(DataType type, Object field, Object literal) {
 
     @Override
     public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal) {
         return compareLiteral(type, literal, min) >= 0;
     }
 

fluss-common/src/main/java/org/apache/fluss/predicate/LessThan.java

@@ -44,7 +44,7 @@ public boolean test(DataType type, Object field, Object literal) {
 
     @Override
     public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal) {
         return compareLiteral(type, literal, min) > 0;
     }
 

fluss-common/src/main/java/org/apache/fluss/predicate/NotEqual.java

@@ -44,7 +44,7 @@ public boolean test(DataType type, Object field, Object literal) {
 
     @Override
     public boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal) {
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal) {
         // ony when max == min == literal, the result is false,
         // otherwise, the row set MAY contain the literal.
         return compareLiteral(type, literal, min) != 0 || compareLiteral(type, literal, max) != 0;

fluss-common/src/main/java/org/apache/fluss/predicate/NotIn.java

@@ -55,9 +55,9 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             List<Object> literals) {
-        if (nullCount != null && rowCount == nullCount) {
+        if (nullCount >= 0 && rowCount == nullCount) {
             return false;
         }
         for (Object literal : literals) {

fluss-common/src/main/java/org/apache/fluss/predicate/NullFalseLeafBinaryFunction.java

@@ -33,7 +33,7 @@ public abstract class NullFalseLeafBinaryFunction extends LeafFunction {
     public abstract boolean test(DataType type, Object field, Object literal);
 
     public abstract boolean test(
-            DataType type, long rowCount, Object min, Object max, Long nullCount, Object literal);
+            DataType type, long rowCount, Object min, Object max, long nullCount, Object literal);
 
     @Override
     public boolean test(DataType type, Object field, List<Object> literals) {
@@ -49,9 +49,9 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             List<Object> literals) {
-        if (nullCount != null) {
+        if (nullCount >= 0) {
             if (rowCount == nullCount || literals.get(0) == null) {
                 return false;
             }

fluss-common/src/main/java/org/apache/fluss/predicate/Or.java

@@ -51,7 +51,7 @@ public boolean test(
             long rowCount,
             InternalRow minValues,
             InternalRow maxValues,
-            Long[] nullCounts,
+            int[] nullCounts,
             List<Predicate> children) {
         for (Predicate child : children) {
             if (child.test(rowCount, minValues, maxValues, nullCounts)) {

fluss-common/src/main/java/org/apache/fluss/predicate/Predicate.java

@@ -45,7 +45,7 @@ public interface Predicate extends Serializable {
      * @return return true is likely to hit (there may also be false positives), return false is
      *     absolutely not possible to hit.
      */
-    boolean test(long rowCount, InternalRow minValues, InternalRow maxValues, Long[] nullCounts);
+    boolean test(long rowCount, InternalRow minValues, InternalRow maxValues, int[] nullCounts);
 
     /**
      * @return the negation predicate of this predicate if possible.

fluss-common/src/main/java/org/apache/fluss/predicate/StartsWith.java

@@ -51,7 +51,7 @@ public boolean test(
             long rowCount,
             Object min,
             Object max,
-            Long nullCount,
+            long nullCount,
             Object patternLiteral) {
         String minStr = min.toString();
         String maxStr = max.toString();

fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatchStatistics.java

@@ -70,7 +70,7 @@ public class DefaultLogRecordBatchStatistics implements LogRecordBatchStatistics
 
     private final int[] statsIndexMapping;
 
-    private final Long[] statsNullCounts;
+    private final int[] statsNullCounts;
 
     // Offsets for min/max values in the byte array
     private final int minValuesOffset;
@@ -80,7 +80,7 @@ public class DefaultLogRecordBatchStatistics implements LogRecordBatchStatistics
 
     private InternalRow cachedMinRow;
     private InternalRow cachedMaxRow;
-    private Long[] cachedNullCounts;
+    private int[] cachedNullCounts;
 
     private final int[] reversedStatsIndexMapping;
 
@@ -91,7 +91,7 @@ public DefaultLogRecordBatchStatistics(
             int size,
             RowType rowType,
             int schemaId,
-            Long[] nullCounts,
+            int[] nullCounts,
             int minValuesOffset,
             int maxValuesOffset,
             int minValuesSize,
@@ -146,16 +146,15 @@ public InternalRow getMaxValues() {
     }
 
     @Override
-    public Long[] getNullCounts() {
+    public int[] getNullCounts() {
         if (cachedNullCounts != null) {
             return cachedNullCounts;
         }
-        cachedNullCounts = new Long[rowType.getFieldCount()];
+        cachedNullCounts = new int[rowType.getFieldCount()];
+        Arrays.fill(cachedNullCounts, LogRecordBatchStatistics.NULL_COUNT_UNAVAILABLE);
         for (int i = 0; i < rowType.getFieldCount(); i++) {
             if (this.reversedStatsIndexMapping[i] >= 0) {
                 cachedNullCounts[i] = statsNullCounts[reversedStatsIndexMapping[i]];
-            } else {
-                cachedNullCounts[i] = -1L;
             }
         }
         return cachedNullCounts;

fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatchStatistics.java

@@ -22,6 +22,9 @@
 /** Statistics information of {@link LogRecordBatch LogRecordBatch}. */
 public interface LogRecordBatchStatistics {
 
+    /** Sentinel value indicating that null count statistics are not available for a field. */
+    int NULL_COUNT_UNAVAILABLE = -1;
+
     /**
      * Get the minimum values as an InternalRow.
      *
@@ -37,11 +40,12 @@ public interface LogRecordBatchStatistics {
     InternalRow getMaxValues();
 
     /**
-     * Get the null counts for each field.
+     * Get the null counts for each field. Uses {@code -1} to indicate that null count statistics
+     * are not available for a particular field.
      *
-     * @return Array of null counts
+     * @return Array of null counts, where -1 means not available
      */
-    Long[] getNullCounts();
+    int[] getNullCounts();
 
     /**
      * Whether the statistics information for a specific field is available.

fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatchStatisticsCollector.java

@@ -47,7 +47,7 @@ public class LogRecordBatchStatisticsCollector {
     // Statistics arrays (only for columns that need statistics)
     private final Object[] minValues;
     private final Object[] maxValues;
-    private final Long[] nullCounts;
+    private final int[] nullCounts;
 
     private final LogRecordBatchStatisticsWriter statisticsWriter;
 
@@ -58,11 +58,11 @@ public LogRecordBatchStatisticsCollector(RowType rowType, int[] statsIndexMappin
         // Initialize statistics arrays
         this.minValues = new Object[statsIndexMapping.length];
         this.maxValues = new Object[statsIndexMapping.length];
-        this.nullCounts = new Long[statsIndexMapping.length];
+        this.nullCounts = new int[statsIndexMapping.length];
 
         this.statisticsWriter = new LogRecordBatchStatisticsWriter(rowType, statsIndexMapping);
 
-        Arrays.fill(nullCounts, 0L);
+        Arrays.fill(nullCounts, 0);
     }
 
     /**
@@ -116,7 +116,7 @@ public int estimatedSizeInBytes() {
 
     /** Reset the collector to collect new statistics. */
     public void reset() {
-        Arrays.fill(nullCounts, 0L);
+        Arrays.fill(nullCounts, 0);
         Arrays.fill(minValues, null);
         Arrays.fill(maxValues, null);
     }

fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatchStatisticsParser.java

@@ -113,9 +113,9 @@ public static DefaultLogRecordBatchStatistics parseStatistics(
 
             // Read null counts at fixed offset
             int nullCountsStart = position + nullCountsOffset(statisticsColumnCount);
-            Long[] nullCounts = new Long[statisticsColumnCount];
+            int[] nullCounts = new int[statisticsColumnCount];
             for (int i = 0; i < statisticsColumnCount; i++) {
-                nullCounts[i] = (long) segment.getInt(nullCountsStart + 4 * i);
+                nullCounts[i] = segment.getInt(nullCountsStart + 4 * i);
             }
 
             // Read min values size at fixed offset