fix: non-batched messages cause sql query to fail (#3684)

pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/MessageParser.java

@@ -86,7 +86,7 @@ public class MessageParser {
 
             if (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch()) {
                 processor.process(
-                        RawMessageImpl.get(refCntMsgMetadata, null, uncompressedPayload, ledgerId, entryId, 0));
+                        RawMessageImpl.get(refCntMsgMetadata, null, uncompressedPayload.retain(), ledgerId, entryId, 0));
             } else {
                 // handle batch message enqueuing; uncompressed payload has all messages in batch
                 receiveIndividualMessagesFromBatch(refCntMsgMetadata, uncompressedPayload, ledgerId, entryId, processor);

tests/integration/src/test/java/org/apache/pulsar/tests/integration/presto/TestBasicPresto.java

@@ -67,7 +67,16 @@ public class TestBasicPresto extends PulsarTestSuite {
     }
 
     @Test
-    public void testSimpleSQLQuery() throws Exception {
+    public void testSimpleSQLQueryBatched() throws Exception {
+        testSimpleSQLQuery(true);
+    }
+
+    @Test
+    public void testSimpleSQLQueryNonBatched() throws Exception {
+        testSimpleSQLQuery(false);
+    }
+    
+    public void testSimpleSQLQuery(boolean isBatched) throws Exception {
 
         @Cleanup
         PulsarClient pulsarClient = PulsarClient.builder()
@@ -79,9 +88,9 @@ public class TestBasicPresto extends PulsarTestSuite {
         @Cleanup
         Producer<Stock> producer = pulsarClient.newProducer(JSONSchema.of(Stock.class))
                 .topic(stocksTopic)
+                .enableBatching(isBatched)
                 .create();
 
-
         for (int i = 0 ; i < NUM_OF_STOCKS; ++i) {
             final Stock stock = new Stock(i,"STOCK_" + i , 100.0 + i * 10);
             producer.send(stock);