[Pulsar sql] Support keyValue Schema add integration test.

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

@@ -22,12 +22,22 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.schema.JSONSchema;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.schema.KeyValue;
+import org.apache.pulsar.common.schema.KeyValueEncodingType;
+import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
+import org.awaitility.Awaitility;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
 @Slf4j
 public class TestBasicPresto extends TestPulsarSQLBase {
 
@@ -57,6 +67,64 @@ public void testSimpleSQLQueryNonBatched() throws Exception {
         pulsarSQLBasicTest(topicName, false, false);
     }
 
+    @DataProvider(name = "keyValueEncodingType")
+    public Object[][] keyValueEncodingType() {
+        return new Object[][] { { KeyValueEncodingType.INLINE }, { KeyValueEncodingType.SEPARATED } };
+    }
+
+    @Test(dataProvider = "keyValueEncodingType")
+    public void testKeyValueSchema(KeyValueEncodingType type) throws Exception {
+        waitPulsarSQLReady();
+        TopicName topicName = TopicName.get("public/default/stocks" + randomName(20));
+        @Cleanup
+        PulsarClient pulsarClient = PulsarClient.builder()
+                .serviceUrl(pulsarCluster.getPlainTextServiceUrl())
+                .build();
+
+        @Cleanup
+        Producer<KeyValue<Stock,Stock>> producer = pulsarClient.newProducer(Schema
+                .KeyValue(Schema.AVRO(Stock.class), Schema.AVRO(Stock.class), type))
+                .topic(topicName.toString())
+                .create();
+
+        for (int i = 0 ; i < NUM_OF_STOCKS; ++i) {
+            int j = 100 * i;
+            final Stock stock1 = new Stock(j, "STOCK_" + j , 100.0 + j * 10);
+            final Stock stock2 = new Stock(i, "STOCK_" + i , 100.0 + i * 10);
+            producer.send(new KeyValue<>(stock1, stock2));
+        }
+
+        producer.flush();
+
+        validateMetadata(topicName);
+
+        Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+                () -> {
+                    ContainerExecResult containerExecResult = execQuery(
+                            String.format("select * from pulsar.\"%s\".\"%s\" order by entryid;",
+                                    topicName.getNamespace(), topicName.getLocalName()));
+                    assertThat(containerExecResult.getExitCode()).isEqualTo(0);
+                    log.info("select sql query output \n{}", containerExecResult.getStdout());
+                    String[] split = containerExecResult.getStdout().split("\n");
+                    assertThat(split.length).isEqualTo(NUM_OF_STOCKS);
+                    String[] split2 = containerExecResult.getStdout().split("\n|,");
+                    for (int i = 0; i < NUM_OF_STOCKS; ++i) {
+                        int j = 100 * i;
+                        assertThat(split2).contains("\"" + i + "\"");
+                        assertThat(split2).contains("\"" + "STOCK_" + i + "\"");
+                        assertThat(split2).contains("\"" + (100.0 + i * 10) + "\"");
+
+                        assertThat(split2).contains("\"" + j + "\"");
+                        assertThat(split2).contains("\"" + "STOCK_" + j + "\"");
+                        assertThat(split2).contains("\"" + (100.0 + j * 10) + "\"");
+                    }
+                }
+        );
+
+    }
+
+
+
     @Override
     protected int prepareData(TopicName topicName, boolean isBatch, boolean useNsOffloadPolices) throws Exception {
         @Cleanup

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

@@ -57,7 +57,7 @@ protected void pulsarSQLBasicTest(TopicName topic, boolean isBatch, boolean useN
         validateData(topic, messageCnt);
     }
 
-    private void waitPulsarSQLReady() throws Exception {
+    public void waitPulsarSQLReady() throws Exception {
         // wait until presto worker started
         ContainerExecResult result;
         do {
@@ -102,7 +102,7 @@ protected int prepareData(TopicName topicName, boolean isBatch, boolean useNsOff
         throw new Exception("Unsupported operation prepareData.");
     }
 
-    private void validateMetadata(TopicName topicName) throws Exception {
+    public void validateMetadata(TopicName topicName) throws Exception {
         ContainerExecResult result = execQuery("show schemas in pulsar;");
         assertThat(result.getExitCode()).isEqualTo(0);
         assertThat(result.getStdout()).contains(topicName.getNamespace());
@@ -122,7 +122,7 @@ private void validateMetadata(TopicName topicName) throws Exception {
         );
     }
 
-    private void validateData(TopicName topicName, int messageNum) throws Exception {
+    public void validateData(TopicName topicName, int messageNum) throws Exception {
         String namespace = topicName.getNamespace();
         String topic = topicName.getLocalName();