[globalindex] Fix NPE in SortedFileMetaSelector when index contains empty string key

paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileMetaSelector.java

@@ -232,16 +232,29 @@ public Optional<List<GlobalIndexIOMeta>> visitNonFieldLeaf(LeafPredicate predica
     }
 
     protected boolean overlaps(SortedIndexFileMeta meta, Object from, Object to) {
-        return comparator.compare(from, deserialize(meta.lastKey())) <= 0
-                && comparator.compare(to, deserialize(meta.firstKey())) >= 0;
+        if (meta.firstKey() != null && comparator.compare(to, deserialize(meta.firstKey())) < 0) {
+            return false;
+        }
+        if (meta.lastKey() != null && comparator.compare(from, deserialize(meta.lastKey())) > 0) {
+            return false;
+        }
+        return true;
     }
 
     protected int compareFirstKey(SortedIndexFileMeta meta, Object literal) {
-        return comparator.compare(deserialize(meta.firstKey()), literal);
+        byte[] firstKey = meta.firstKey();
+        if (firstKey == null) {
+            return -1;
+        }
+        return comparator.compare(deserialize(firstKey), literal);
     }
 
     protected int compareLastKey(SortedIndexFileMeta meta, Object literal) {
-        return comparator.compare(deserialize(meta.lastKey()), literal);
+        byte[] lastKey = meta.lastKey();
+        if (lastKey == null) {
+            return 1;
+        }
+        return comparator.compare(deserialize(lastKey), literal);
     }
 
     protected int compareKeys(Object left, Object right) {

paimon-common/src/test/java/org/apache/paimon/globalindex/SortedFileMetaSelectorTest.java

@@ -277,6 +277,65 @@ private BinaryString str(String value) {
         return BinaryString.fromString(value);
     }
 
+    @Test
+    public void testEmptyStringKeyDoesNotThrowNPE() {
+        KeySerializer serializer = KeySerializer.create(new VarCharType());
+        FieldRef ref = new FieldRef(1, "page_host", new VarCharType());
+
+        byte[] emptyKey = serializer.serialize(BinaryString.EMPTY_UTF8);
+        byte[] normalKey = serializer.serialize(BinaryString.fromString("www.example.com"));
+
+        SortedIndexFileMeta metaWithEmptyFirstKey =
+                new SortedIndexFileMeta(emptyKey, normalKey, false);
+        SortedIndexFileMeta metaWithNormalKeys =
+                new SortedIndexFileMeta(
+                        serializer.serialize(BinaryString.fromString("aaa.com")),
+                        serializer.serialize(BinaryString.fromString("zzz.com")),
+                        false);
+        SortedIndexFileMeta metaOnlyNulls = new SortedIndexFileMeta(null, null, true);
+
+        List<GlobalIndexIOMeta> testFiles =
+                Arrays.asList(
+                        new GlobalIndexIOMeta(
+                                new Path("file_empty"), 1, metaWithEmptyFirstKey.serialize()),
+                        new GlobalIndexIOMeta(
+                                new Path("file_normal"), 1, metaWithNormalKeys.serialize()),
+                        new GlobalIndexIOMeta(
+                                new Path("file_nulls"), 1, metaOnlyNulls.serialize()));
+
+        SortedFileMetaSelector selector = new SortedFileMetaSelector(testFiles, serializer);
+
+        // visitEqual should not throw NPE
+        Optional<List<GlobalIndexIOMeta>> result =
+                selector.visitEqual(ref, BinaryString.fromString("www.example.com"));
+        Assertions.assertThat(result).isNotEmpty();
+        assertFiles(result.get(), Arrays.asList("file_empty", "file_normal"));
+
+        // visitLessThan should not throw NPE
+        result = selector.visitLessThan(ref, BinaryString.fromString("bbb.com"));
+        Assertions.assertThat(result).isNotEmpty();
+        assertFiles(result.get(), Arrays.asList("file_empty", "file_normal"));
+
+        // visitGreaterThan should not throw NPE
+        result = selector.visitGreaterThan(ref, BinaryString.fromString("www.example.com"));
+        Assertions.assertThat(result).isNotEmpty();
+
+        // visitIn should not throw NPE
+        result =
+                selector.visitIn(
+                        ref,
+                        Arrays.asList(
+                                BinaryString.fromString("www.example.com"),
+                                BinaryString.fromString("zzz.com")));
+        Assertions.assertThat(result).isNotEmpty();
+
+        // visitBetween should not throw NPE
+        result =
+                selector.visitBetween(
+                        ref, BinaryString.EMPTY_UTF8, BinaryString.fromString("zzz.com"));
+        Assertions.assertThat(result).isNotEmpty();
+    }
+
     private static class LengthPrefixedStringSerializer implements KeySerializer {
 
         @Override