[common] Collect min and max stats for binary columns

paimon-common/src/main/java/org/apache/paimon/format/SimpleColStats.java

@@ -20,6 +20,7 @@
 
 import javax.annotation.Nullable;
 
+import java.util.Arrays;
 import java.util.Objects;
 
 /**
@@ -70,18 +71,25 @@ public boolean equals(Object o) {
             return false;
         }
         SimpleColStats that = (SimpleColStats) o;
-        return Objects.equals(min, that.min)
-                && Objects.equals(max, that.max)
+        return Objects.deepEquals(min, that.min)
+                && Objects.deepEquals(max, that.max)
                 && Objects.equals(nullCount, that.nullCount);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(min, max, nullCount);
+        return Objects.hash(
+                min instanceof byte[] ? Arrays.hashCode((byte[]) min) : min,
+                max instanceof byte[] ? Arrays.hashCode((byte[]) max) : max,
+                nullCount);
     }
 
     @Override
     public String toString() {
-        return String.format("{%s, %s, %d}", min, max, nullCount);
+        return String.format(
+                "{%s, %s, %d}",
+                min instanceof byte[] ? Arrays.toString((byte[]) min) : min,
+                max instanceof byte[] ? Arrays.toString((byte[]) max) : max,
+                nullCount);
     }
 }

paimon-common/src/main/java/org/apache/paimon/statistics/FullSimpleColStatsCollector.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.data.serializer.Serializer;
 import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.utils.SortUtil;
 
 /** The full stats collector which will report null count, min value, max value if available. */
 public class FullSimpleColStatsCollector extends AbstractSimpleColStatsCollector {
@@ -31,7 +32,17 @@ public void collect(Object field, Serializer<Object> fieldSerializer) {
             return;
         }
 
-        // TODO use comparator for not comparable types and extract this logic to a util class
+        if (field instanceof byte[]) {
+            byte[] b = (byte[]) field;
+            if (minValue == null || SortUtil.compareBinary(b, (byte[]) minValue) < 0) {
+                minValue = fieldSerializer.copy(field);
+            }
+            if (maxValue == null || SortUtil.compareBinary(b, (byte[]) maxValue) > 0) {
+                maxValue = fieldSerializer.copy(field);
+            }
+            return;
+        }
+
         if (!(field instanceof Comparable)) {
             return;
         }

paimon-common/src/main/java/org/apache/paimon/statistics/TruncateSimpleColStatsCollector.java

@@ -22,12 +22,14 @@
 import org.apache.paimon.data.serializer.Serializer;
 import org.apache.paimon.format.SimpleColStats;
 import org.apache.paimon.utils.Preconditions;
+import org.apache.paimon.utils.SortUtil;
 
+import java.util.Arrays;
 import java.util.regex.Pattern;
 
 /**
  * The truncate stats collector which will report null count, truncated min/max value. Currently,
- * truncation only performs on the {@link BinaryString} value.
+ * truncation only performs on the {@link BinaryString} and {@code byte[]} values.
  */
 public class TruncateSimpleColStatsCollector extends AbstractSimpleColStatsCollector {
 
@@ -58,7 +60,26 @@ public void collect(Object field, Serializer<Object> fieldSerializer) {
             return;
         }
 
-        // TODO use comparator for not comparable types and extract this logic to a util class
+        if (field instanceof byte[]) {
+            byte[] bytes = (byte[]) field;
+            if (minValue == null || SortUtil.compareBinary(bytes, (byte[]) minValue) < 0) {
+                minValue = fieldSerializer.copy(truncateMin(field));
+            }
+            if (maxValue == null || SortUtil.compareBinary(bytes, (byte[]) maxValue) > 0) {
+                Object max = truncateMax(field);
+                // may fail
+                if (max != null) {
+                    if (max != field) {
+                        // copied in `truncateMax`
+                        maxValue = max;
+                    } else {
+                        maxValue = fieldSerializer.copy(max);
+                    }
+                }
+            }
+            return;
+        }
+
         if (!(field instanceof Comparable)) {
             return;
         }
@@ -106,6 +127,9 @@ private Object truncateMin(Object field) {
         }
         if (field instanceof BinaryString) {
             return ((BinaryString) field).substring(0, length);
+        } else if (field instanceof byte[]) {
+            byte[] bytes = (byte[]) field;
+            return bytes.length <= length ? bytes : Arrays.copyOf(bytes, length);
         } else {
             return field;
         }
@@ -143,6 +167,26 @@ private Object truncateMax(Object field) {
             }
             failed = true;
             return null; // Cannot find a valid upper bound
+        } else if (field instanceof byte[]) {
+            byte[] bytes = (byte[]) field;
+
+            // No need to increment if the input length is under the truncate length
+            if (bytes.length <= length) {
+                return field;
+            }
+
+            byte[] truncated = Arrays.copyOf(bytes, length);
+
+            // Try incrementing the bytes from the end
+            for (int i = length - 1; i >= 0; i--) {
+                // No overflow
+                if (truncated[i] != (byte) 0xFF) {
+                    truncated[i]++;
+                    return Arrays.copyOf(truncated, i + 1);
+                }
+            }
+            failed = true;
+            return null; // Cannot find a valid upper bound
         } else {
             return field;
         }

paimon-common/src/test/java/org/apache/paimon/statistics/SimpleColStatsCollectorTest.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.types.DataField;
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarBinaryType;
 import org.apache.paimon.types.VarCharType;
 import org.apache.paimon.utils.StringUtils;
 
@@ -193,6 +194,60 @@ public void testFullCopied() {
         assertThat(stats.max()).isNotSameAs(str);
     }
 
+    @Test
+    public void testFullBinaryMinMax() {
+        Serializer<Object> serializer =
+                (Serializer) InternalSerializers.create(new VarBinaryType());
+        FullSimpleColStatsCollector collector = new FullSimpleColStatsCollector();
+        collector.collect(new byte[] {1, 2, 3}, serializer);
+        collector.collect(new byte[] {(byte) 200}, serializer);
+        collector.collect(new byte[] {0}, serializer);
+        collector.collect(null, serializer);
+
+        SimpleColStats stats = collector.result();
+        assertThat((byte[]) stats.min()).isEqualTo(new byte[] {0});
+        assertThat((byte[]) stats.max()).isEqualTo(new byte[] {(byte) 200});
+        assertThat(stats.nullCount()).isEqualTo(1L);
+    }
+
+    @Test
+    public void testTruncateBinaryMinMax() {
+        Serializer<Object> serializer =
+                (Serializer) InternalSerializers.create(new VarBinaryType());
+        TruncateSimpleColStatsCollector collector = new TruncateSimpleColStatsCollector(2);
+        collector.collect(new byte[] {5, 7, 9, 11}, serializer);
+        collector.collect(new byte[] {5, 7, (byte) 200}, serializer);
+        collector.collect(null, serializer);
+
+        SimpleColStats stats = collector.result();
+        assertThat((byte[]) stats.min()).isEqualTo(new byte[] {5, 7});
+        assertThat((byte[]) stats.max()).isEqualTo(new byte[] {5, 8});
+        assertThat(stats.nullCount()).isEqualTo(1L);
+
+        stats =
+                collector.convert(
+                        new SimpleColStats(new byte[] {1, 2, 3}, new byte[] {1, 2, 3}, 0L));
+        assertThat((byte[]) stats.min()).isEqualTo(new byte[] {1, 2});
+        assertThat((byte[]) stats.max()).isEqualTo(new byte[] {1, 3});
+    }
+
+    @Test
+    public void testTruncateBinaryFail() {
+        TruncateSimpleColStatsCollector collector = new TruncateSimpleColStatsCollector(2);
+        Serializer<Object> serializer =
+                (Serializer) InternalSerializers.create(new VarBinaryType());
+        byte[] bytes = new byte[] {(byte) 0xFF, (byte) 0xFF, 7};
+
+        collector.collect(bytes, serializer);
+        SimpleColStats stats = collector.result();
+        assertThat(stats.min()).isNull();
+        assertThat(stats.max()).isNull();
+
+        stats = collector.convert(new SimpleColStats(bytes, bytes, 0L));
+        assertThat(stats.min()).isNull();
+        assertThat(stats.max()).isNull();
+    }
+
     @Test
     public void testTruncateFail() {
         TruncateSimpleColStatsCollector collector = new TruncateSimpleColStatsCollector(3);

paimon-core/src/test/java/org/apache/paimon/append/AppendOnlyWriterTest.java

@@ -62,6 +62,7 @@
 import org.apache.paimon.types.DataTypes;
 import org.apache.paimon.types.IntType;
 import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.VarBinaryType;
 import org.apache.paimon.types.VarCharType;
 import org.apache.paimon.utils.CommitIncrement;
 import org.apache.paimon.utils.ExecutorThreadFactory;
@@ -170,6 +171,48 @@ public void testSingleWrite() throws Exception {
                 .isEqualTo(CoreOptions.FILE_FORMAT_AVRO);
     }
 
+    @Test
+    public void testBinaryColumnStatsRoundTrip() throws Exception {
+        RowType binarySchema =
+                RowType.builder()
+                        .fields(
+                                new DataType[] {new IntType(), new VarBinaryType()},
+                                new String[] {"id", "data"})
+                        .build();
+        Map<String, String> options = new HashMap<>();
+        options.put("metadata.stats-mode", "full");
+
+        AppendOnlyWriter writer =
+                createWriterBase(
+                                1024 * 1024L,
+                                null,
+                                binarySchema,
+                                false,
+                                true,
+                                true,
+                                true,
+                                Collections.emptyList(),
+                                compactBefore -> Collections.emptyList(),
+                                options)
+                        .getKey();
+
+        writer.write(GenericRow.of(1, new byte[] {1, 2, 3}));
+        writer.write(GenericRow.of(2, new byte[] {(byte) 200}));
+        writer.write(GenericRow.of(3, new byte[] {0}));
+        CommitIncrement increment = writer.prepareCommit(true);
+        writer.close();
+
+        DataFileMeta meta = increment.newFilesIncrement().newFiles().get(0);
+        assertThat(LocalFileIO.create().exists(pathFactory.toPath(meta))).isTrue();
+
+        assertThat(meta.valueStats().minValues().isNullAt(1))
+                .as("binary column min must be present in file stats")
+                .isFalse();
+        assertThat(meta.valueStats().minValues().getBinary(1)).isEqualTo(new byte[] {0});
+        assertThat(meta.valueStats().maxValues().getBinary(1)).isEqualTo(new byte[] {(byte) 200});
+        assertThat(meta.valueStats().nullCounts().getLong(1)).isEqualTo(0L);
+    }
+
     @Test
     public void testMultipleCommits() throws Exception {
         RecordWriter<InternalRow> writer =

paimon-core/src/test/java/org/apache/paimon/table/ColumnTypeFileMetaTestBase.java

@@ -91,13 +91,10 @@ protected void validateStatsField(List<DataFileMeta> fileMetaList) {
         for (DataFileMeta fileMeta : fileMetaList) {
             SimpleStats stats = getTableValueStats(fileMeta);
             assertThat(stats.minValues().getFieldCount()).isEqualTo(12);
-            for (int i = 0; i < 11; i++) {
+            for (int i = 0; i < 12; i++) {
                 assertThat(stats.minValues().isNullAt(i)).isFalse();
                 assertThat(stats.maxValues().isNullAt(i)).isFalse();
             }
-            // Min and max value of binary type is null
-            assertThat(stats.minValues().isNullAt(11)).isTrue();
-            assertThat(stats.maxValues().isNullAt(11)).isTrue();
         }
     }
 

paimon-format/src/main/java/org/apache/paimon/format/parquet/ParquetSimpleStatsExtractor.java

@@ -142,6 +142,16 @@ private SimpleColStats toFieldStats(
                                 BinaryString.fromString(stringStats.maxAsString()),
                                 nullCount);
                 break;
+            case BINARY:
+            case VARBINARY:
+                assertStatsClass(field, stats, BinaryStatistics.class);
+                BinaryStatistics binaryStats = (BinaryStatistics) stats;
+                fieldStats =
+                        new SimpleColStats(
+                                binaryStats.genericGetMin().getBytes(),
+                                binaryStats.genericGetMax().getBytes(),
+                                nullCount);
+                break;
             case BOOLEAN:
                 assertStatsClass(field, stats, BooleanStatistics.class);
                 BooleanStatistics boolStats = (BooleanStatistics) stats;

paimon-format/src/test/java/org/apache/paimon/format/orc/OrcSimpleStatsExtractorTest.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.format.orc;
 
 import org.apache.paimon.format.FileFormat;
+import org.apache.paimon.format.SimpleColStats;
 import org.apache.paimon.format.SimpleColStatsExtractorTest;
 import org.apache.paimon.format.orc.filter.OrcSimpleStatsExtractor;
 import org.apache.paimon.options.Options;
@@ -27,6 +28,7 @@
 import org.apache.paimon.types.BinaryType;
 import org.apache.paimon.types.BooleanType;
 import org.apache.paimon.types.CharType;
+import org.apache.paimon.types.DataType;
 import org.apache.paimon.types.DateType;
 import org.apache.paimon.types.DecimalType;
 import org.apache.paimon.types.DoubleType;
@@ -80,6 +82,16 @@ protected RowType rowType() {
                 .build();
     }
 
+    @Override
+    protected SimpleColStats regenerate(SimpleColStats stats, DataType type) {
+        switch (type.getTypeRoot()) {
+            case BINARY:
+            case VARBINARY:
+                return new SimpleColStats(null, null, stats.nullCount());
+        }
+        return stats;
+    }
+
     @Override
     protected String fileCompression() {
         return "LZ4";