[spark] Fix reported statistics does not do column pruning (#4137)

Author: ulysses-you

paimon-common/src/main/java/org/apache/paimon/types/ArrayType.java

@@ -57,6 +57,11 @@ public DataType getElementType() {
         return elementType;
     }
 
+    @Override
+    public int defaultSize() {
+        return elementType.defaultSize();
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new ArrayType(isNullable, elementType.copy());

paimon-common/src/main/java/org/apache/paimon/types/BigIntType.java

@@ -41,6 +41,11 @@ public BigIntType() {
         this(true);
     }
 
+    @Override
+    public int defaultSize() {
+        return 8;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new BigIntType(isNullable);

paimon-common/src/main/java/org/apache/paimon/types/BinaryType.java

@@ -65,6 +65,11 @@ public int getLength() {
         return length;
     }
 
+    @Override
+    public int defaultSize() {
+        return length;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new BinaryType(isNullable, length);

paimon-common/src/main/java/org/apache/paimon/types/BooleanType.java

@@ -40,6 +40,11 @@ public BooleanType() {
         this(true);
     }
 
+    @Override
+    public int defaultSize() {
+        return 1;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new BooleanType(isNullable);

paimon-common/src/main/java/org/apache/paimon/types/CharType.java

@@ -67,6 +67,11 @@ public int getLength() {
         return length;
     }
 
+    @Override
+    public int defaultSize() {
+        return length;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new CharType(isNullable, length);

paimon-common/src/main/java/org/apache/paimon/types/DataType.java

@@ -104,6 +104,9 @@ public boolean is(DataTypeFamily family) {
         return typeRoot.getFamilies().contains(family);
     }
 
+    /** The default size of a value of this data type, used internally for size estimation. */
+    public abstract int defaultSize();
+
     /**
      * Returns a deep copy of this type with possibly different nullability.
      *

paimon-common/src/main/java/org/apache/paimon/types/DateType.java

@@ -44,6 +44,11 @@ public DateType() {
         this(true);
     }
 
+    @Override
+    public int defaultSize() {
+        return 4;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new DateType(isNullable);

paimon-common/src/main/java/org/apache/paimon/types/DecimalType.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.types;
 
 import org.apache.paimon.annotation.Public;
+import org.apache.paimon.data.Decimal;
 
 import java.util.Objects;
 
@@ -86,6 +87,11 @@ public int getScale() {
         return scale;
     }
 
+    @Override
+    public int defaultSize() {
+        return Decimal.isCompact(precision) ? 8 : 16;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new DecimalType(isNullable, precision, scale);

paimon-common/src/main/java/org/apache/paimon/types/DoubleType.java

@@ -40,6 +40,11 @@ public DoubleType() {
         this(true);
     }
 
+    @Override
+    public int defaultSize() {
+        return 8;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new DoubleType(isNullable);

paimon-common/src/main/java/org/apache/paimon/types/FloatType.java

@@ -41,6 +41,11 @@ public FloatType() {
         this(true);
     }
 
+    @Override
+    public int defaultSize() {
+        return 4;
+    }
+
     @Override
     public DataType copy(boolean isNullable) {
         return new FloatType(isNullable);