order book transformation test

Author: jerrinot

connector/src/test/java/io/questdb/kafka/QuestDBSinkConnectorEmbeddedTest.java

@@ -2789,6 +2789,75 @@ public void testOrderBookToArraySMT_targetCollidesWithExisting() {
         );
     }
 
+    @Test
+    public void testMarketData_orderBookToArray_withTimestamp_schemaless() {
+        connect.kafka().createTopic(topicName, 1);
+        Map<String, String> props = ConnectTestUtils.baseConnectorProps(questDBContainer, topicName, true);
+        props.put("value.converter.schemas.enable", "false");
+        props.put(QuestDBSinkConnectorConfig.DESIGNATED_TIMESTAMP_COLUMN_NAME_CONFIG, "received_at");
+        props.put(QuestDBSinkConnectorConfig.INCLUDE_KEY_CONFIG, "false");
+        props.put(QuestDBSinkConnectorConfig.SYMBOL_COLUMNS_CONFIG, "market");
+
+        // The connector parses received_at natively; U+ accepts variable-width fractional seconds (1-6 digits)
+        props.put(QuestDBSinkConnectorConfig.TIMESTAMP_STRING_FIELDS, "received_at");
+        props.put(QuestDBSinkConnectorConfig.TIMESTAMP_FORMAT, "yyyy-MM-ddTHH:mm:ss.U+Z");
+
+        // Transform pipeline:
+        // 1. ExtractField unwraps the schema/payload envelope
+        // 2. OrderBookToArray transposes bids/asks arrays-of-objects into double[][]
+        props.put("transforms", "extractPayload,orderbook");
+        props.put("transforms.extractPayload.type", "org.apache.kafka.connect.transforms.ExtractField$Value");
+        props.put("transforms.extractPayload.field", "payload");
+        props.put("transforms.orderbook.type", "io.questdb.kafka.OrderBookToArray$Value");
+        props.put("transforms.orderbook.mappings", "bids:bids:price,amount;asks:asks:price,amount");
+
+        connect.configureConnector(ConnectTestUtils.CONNECTOR_NAME, props);
+        ConnectTestUtils.assertConnectorTaskRunningEventually(connect);
+
+        // Pre-create QuestDB table: market as symbol, venue as int, bids/asks as 2D double arrays,
+        // received_at as designated timestamp partitioned by hour
+        QuestDBUtils.assertSql(
+                "{\"ddl\":\"OK\"}",
+                "create table " + topicName
+                        + " (market symbol, venue int, bids double[][], asks double[][], received_at timestamp)"
+                        + " timestamp(received_at) partition by hour wal",
+                httpPort,
+                QuestDBUtils.Endpoint.EXEC);
+
+        // Full JSON with schema/payload envelope, matching the real upstream message structure.
+        // price/amount are numeric (not string-encoded) so OrderBookToArray can cast them to double.
+        // Extra fields (side, quote_entry_id, time, type, entry_id) inside bids/asks are ignored
+        // by OrderBookToArray — only the mapped fields (price, amount) are extracted.
+        String json = "{"
+                + "\"schema\":\"market_data\","
+                + "\"payload\":{"
+                +   "\"market\":\"BTCUSD\","
+                +   "\"venue\":1,"
+                +   "\"received_at\":\"2026-01-13T22:00:00.014Z\","
+                +   "\"type\":\"SNAPSHOT\","
+                +   "\"bids\":["
+                +     "{\"side\":\"BUY\",\"amount\":2.45,\"price\":45120.50,\"quote_entry_id\":\"uuid-1\",\"time\":\"2026-01-13T22:00:00.014Z\",\"type\":\"NEW\",\"entry_id\":\"45120.50\"},"
+                +     "{\"side\":\"BUY\",\"amount\":5.12,\"price\":45119.00,\"quote_entry_id\":\"uuid-2\",\"time\":\"2026-01-13T22:00:00.014Z\",\"type\":\"NEW\",\"entry_id\":\"45119.00\"}"
+                +   "],"
+                +   "\"asks\":["
+                +     "{\"side\":\"SELL\",\"amount\":1.83,\"price\":45121.00,\"quote_entry_id\":\"uuid-3\",\"time\":\"2026-01-13T22:00:00.014Z\",\"type\":\"NEW\",\"entry_id\":\"45121.00\"},"
+                +     "{\"side\":\"SELL\",\"amount\":3.27,\"price\":45122.50,\"quote_entry_id\":\"uuid-4\",\"time\":\"2026-01-13T22:00:00.014Z\",\"type\":\"NEW\",\"entry_id\":\"45122.50\"}"
+                +   "]"
+                + "}}";
+        connect.kafka().produce(topicName, json);
+
+        // After OrderBookToArray transpose with mappings "bids:bids:price,amount":
+        // bids: [[45120.5, 45119.0], [2.45, 5.12]]   (row 0 = prices, row 1 = amounts)
+        // asks: [[45121.0, 45122.5], [1.83, 3.27]]
+        // The extra top-level "type" field ("SNAPSHOT") becomes an additional string column in QuestDB
+        QuestDBUtils.assertSqlEventually(
+                "\"market\",\"venue\",\"bids\",\"asks\",\"received_at\"\r\n"
+                        + "\"BTCUSD\",1,\"[[45120.5,45119.0],[2.45,5.12]]\",\"[[45121.0,45122.5],[1.83,3.27]]\",\"2026-01-13T22:00:00.014000Z\"\r\n",
+                "select market, venue, bids, asks, received_at from " + topicName,
+                httpPort
+        );
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testComposedTimestamp_schemaless(boolean useHttp) {