fix: bugfix for predicate batch reader not check predicate nullptr

src/paimon/common/reader/predicate_batch_reader.cpp

@@ -48,6 +48,9 @@ PredicateBatchReader::PredicateBatchReader(std::unique_ptr<BatchReader>&& reader
 Result<std::unique_ptr<PredicateBatchReader>> PredicateBatchReader::Create(
     std::unique_ptr<BatchReader>&& reader, const std::shared_ptr<Predicate>& predicate,
     const std::shared_ptr<MemoryPool>& pool) {
+    if (!predicate) {
+        return Status::Invalid("create predicate batch reader failed. predicate is nullptr");
+    }
     auto predicate_filter = std::dynamic_pointer_cast<PredicateFilter>(predicate);
     if (!predicate_filter) {
         return Status::Invalid(

src/paimon/common/reader/predicate_batch_reader_test.cpp

@@ -188,4 +188,11 @@ TEST_F(PredicateBatchReaderTest, TestFullAndEmptyCase) {
     }
 }
 
+TEST_F(PredicateBatchReaderTest, TestInvalidInput) {
+    auto data_array = PrepareArray(8);
+    auto reader = std::make_unique<MockFileBatchReader>(data_array, data_type_, /*batch_size=*/10);
+    ASSERT_NOK_WITH_MSG(PredicateBatchReader::Create(std::move(reader), nullptr, GetDefaultPool()),
+                        "create predicate batch reader failed. predicate is nullptr");
+}
+
 }  // namespace paimon::test

src/paimon/core/operation/abstract_split_read.cpp

@@ -123,7 +123,7 @@ std::unordered_map<std::string, DeletionFile> AbstractSplitRead::CreateDeletionF
 
 Result<std::unique_ptr<BatchReader>> AbstractSplitRead::ApplyPredicateFilterIfNeeded(
     std::unique_ptr<BatchReader>&& reader, const std::shared_ptr<Predicate>& predicate) const {
-    if (!context_->EnablePredicateFilter()) {
+    if (!context_->EnablePredicateFilter() || predicate == nullptr) {
         return std::move(reader);
     }
     return PredicateBatchReader::Create(std::move(reader), predicate, pool_);