[HTTP-34] Add support for lookup on Row type (#35)

* [HTTP-34] - Adding support for Lookup on Row type and nested types.

This change adds support for SQL lookup joins scenarios where ROW type column is used in JOIN condition.
In that case, Lookup connector converts used columns (including nested types like Arrays, Maps Lists and ROWs) into proper http query:

for GET requests it converts all fields (including ROW columns) to key=value pairs and use them in HTTP query parameters
For Body based requests like POST/PUT it converts entire lookup row into JSON object and use it as query's body.
By default Flink's Json-Format is used for RowData -> Json conversion.

Signed-off-by: Krzysztof Chmielewski <krzysiek.chmielewski@gmail.com>

* Address review comments: fix typos

Signed-off-by: Krzysztof Chmielewski <krzysiek.chmielewski@gmail.com>
Co-authored-by: Andrzej Swatowski <andrzejswatowski@outlook.com>

Author: kristoffSC

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+- Add to Lookup Source support for performing lookup on columns with complex types such as ROW, Map etc.
+- Add support for custom Json Serialization format for SQL Lookup Source when using [GenericJsonQueryCreator](src/main/java/com/getindata/connectors/http/internal/table/lookup/querycreators/GenericJsonQueryCreator.java)
+  The custom format can be defined using Flink's Factory mechanism. The format name can be defined using
+  `lookup-request.format` option. The default format is `json` which means that connector will use FLink's [json-format](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/) 
+
 ## [0.6.0] - 2022-10-05
 
 ### Added

README.md

@@ -133,8 +133,26 @@ the Http Lookup Source DDL property field `gid.connector.http.source.lookup.quer
 
 A default implementation that builds an "ordinary" GET query, i.e. adds `?joinColumn1=value1&joinColumn2=value2&...`
 to the URI of the endpoint,
+
+For body based queries such as POST/PUT requests, the 
 ([GenericGetQueryCreator](src/main/java/com/getindata/connectors/http/internal/table/lookup/querycreators/GenericGetQueryCreator.java))
-is provided and set as a default.
+is provided as a default query creator. This implementation uses Flink's [json-format](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/)  to convert RowData object into Json String.
+
+The important thing worth knowing is that `GenericGetQueryCreator` allows for using custom formats that will perform serialization to Json. Thanks to this, users can create their own logic for converting RowData to Json Strings that will match their requirements
+and use it in HTTP Lookup connector and SQL queries.
+To create a custom format user has to implement Flink's `SerializationSchema` and `SerializationFormatFactory` interfaces and register custom format factory along other factories in
+`resources/META-INF.services/org.apache.flink.table.factories.Factory` file. This is common Flink mechanism for providing custom implementations for various factories.
+In order to use custom format, user has to specify option `'lookup-request.format' = 'customFormatName'`, where `customFormatName` is the identifier of our custom format factory.
+
+Additionally, it is possible to pass query format options from table's DDL.
+This can be done by using option like so: `'lookup-request.format.customFormatName.customFormatProperty' = 'propertyValue'`, for example
+`'lookup-request.format.customFormatName.fail-on-missing-field' = 'true'`. It is important that `customFormatName` part must match `SerializationFormatFactory` identifier used for custom format implementation.
+In this case, the `fail-on-missing-field` will be passed to `SerializationFormatFactory::createEncodingFormat(
+DynamicTableFactory.Context context, ReadableConfig formatOptions)` method in `ReadableConfig` object.
+
+In configuration when default, Flink-Json format is used for `GenericGetQueryCreator`, all options defined in [json-format](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/formats/json/)
+can be passed through table DDL. For example `'lookup-request.format.json.fail-on-missing-field' = 'true'`. In this case, format identifier is `json`.
+
 
 ### HTTP Sink
 The following example shows the minimum Table API example to create a [HttpDynamicSink](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicSink.java) that writes JSON values to an HTTP endpoint using POST method, assuming Flink has JAR of [JSON serializer](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/json/) installed:

src/main/java/com/getindata/connectors/http/LookupArg.java

@@ -4,19 +4,19 @@
 import lombok.RequiredArgsConstructor;
 
 /**
- * Transfer object that contains Http request argument and value.
+ * Transfer object that contains single lookup argument (column name) and its value.
  */
 @Data
 @RequiredArgsConstructor
 public class LookupArg {
 
     /**
-     * HTTP request argument's name.
+     * Lookup argument name.
      */
     private final String argName;
 
     /**
-     * HTTP request argument's value.
+     * Lookup argument value.
      */
     private final String argValue;
 }

src/main/java/com/getindata/connectors/http/LookupQueryCreator.java

@@ -1,7 +1,8 @@
 package com.getindata.connectors.http;
 
 import java.io.Serializable;
-import java.util.List;
+
+import org.apache.flink.table.data.RowData;
 
 /**
  * An interface for a creator of a lookup query in the Http Lookup Source (e.g., the query that
@@ -16,8 +17,8 @@ public interface LookupQueryCreator extends Serializable {
      * Create a lookup query (like the query appended to path in GET request)
      * out of the provided arguments.
      *
-     * @param params the list of {@link LookupArg} containing request parameters.
+     * @param lookupDataRow a {@link RowData} containing request parameters.
      * @return a lookup query.
      */
-    String createLookupQuery(List<LookupArg> params);
+    String createLookupQuery(RowData lookupDataRow);
 }

src/main/java/com/getindata/connectors/http/LookupQueryCreatorFactory.java

@@ -1,8 +1,12 @@
 package com.getindata.connectors.http;
 
+import java.io.Serializable;
+
+import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.table.factories.Factory;
 
 import com.getindata.connectors.http.internal.table.lookup.HttpLookupTableSource;
+import com.getindata.connectors.http.internal.table.lookup.LookupRow;
 
 /**
  * The {@link Factory} that dynamically creates and injects {@link LookupQueryCreator} to
@@ -29,10 +33,12 @@
  * )
  * }</pre>
  */
-public interface LookupQueryCreatorFactory extends Factory {
+public interface LookupQueryCreatorFactory extends Factory, Serializable {
 
     /**
      * @return {@link LookupQueryCreator} custom lookup query creator instance
      */
-    LookupQueryCreator createLookupQueryCreator();
+    LookupQueryCreator createLookupQueryCreator(
+        ReadableConfig readableConfig,
+        LookupRow lookupRow);
 }

src/main/java/com/getindata/connectors/http/internal/PollingClient.java

@@ -1,9 +1,8 @@
 package com.getindata.connectors.http.internal;
 
-import java.util.List;
 import java.util.Optional;
 
-import com.getindata.connectors.http.LookupArg;
+import org.apache.flink.table.data.RowData;
 
 /**
  * A client that is used to get enrichment data from external component.
@@ -12,8 +11,8 @@ public interface PollingClient<T> {
 
     /**
      * Gets enrichment data from external component using provided lookup arguments.
-     * @param lookupArgs The list of {@link LookupArg} containing request parameters.
+     * @param lookupRow A {@link RowData} containing request parameters.
      * @return an optional result of data lookup.
      */
-    Optional<T> pull(List<LookupArg> lookupArgs);
+    Optional<T> pull(RowData lookupRow);
 }

src/main/java/com/getindata/connectors/http/internal/table/lookup/AsyncHttpTableLookupFunction.java

@@ -14,7 +14,6 @@
 import org.apache.flink.table.functions.FunctionContext;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
-import com.getindata.connectors.http.internal.table.lookup.HttpTableLookupFunction.ColumnData;
 import com.getindata.connectors.http.internal.utils.ThreadUtils;
 
 @Slf4j
@@ -81,8 +80,8 @@ public void eval(CompletableFuture<Collection<RowData>> resultFuture, Object...
             publishingThreadPool);
     }
 
-    public ColumnData getColumnData() {
-        return decorate.getColumnData();
+    public LookupRow getLookupRow() {
+        return decorate.getLookupRow();
     }
 
     public HttpLookupConfig getOptions() {

src/main/java/com/getindata/connectors/http/internal/table/lookup/BodyBasedRequestFactory.java

@@ -7,6 +7,9 @@
 import java.net.http.HttpRequest.Builder;
 import java.time.Duration;
 
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+
 import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.utils.uri.URIBuilder;
@@ -15,6 +18,7 @@
  * Implementation of {@link HttpRequestFactory} for REST calls that sends their parameters using
  * request body.
  */
+@Slf4j
 public class BodyBasedRequestFactory extends RequestFactoryBase {
 
     private final String methodName;
@@ -44,6 +48,11 @@ protected Builder setUpRequestMethod(String lookupQuery) {
             .timeout(Duration.ofMinutes(2));
     }
 
+    @Override
+    protected Logger getLogger() {
+        return log;
+    }
+
     private URI constructGetUri() {
         try {
             return new URIBuilder(baseUrl).build();

src/main/java/com/getindata/connectors/http/internal/table/lookup/GetRequestFactory.java

@@ -6,13 +6,17 @@
 import java.net.http.HttpRequest.Builder;
 import java.time.Duration;
 
+import lombok.extern.slf4j.Slf4j;
+import org.slf4j.Logger;
+
 import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.utils.uri.URIBuilder;
 
 /**
  * Implementation of {@link HttpRequestFactory} for GET REST calls.
  */
+@Slf4j
 public class GetRequestFactory extends RequestFactoryBase {
 
     public GetRequestFactory(
@@ -23,6 +27,11 @@ public GetRequestFactory(
         super(lookupQueryCreator, headerPreprocessor, options);
     }
 
+    @Override
+    protected Logger getLogger() {
+        return log;
+    }
+
     /**
      * Method for preparing {@link HttpRequest.Builder} for REST GET request, where lookupQuery
      * is used as query parameters for example:

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupConfig.java

@@ -1,27 +1,29 @@
 package com.getindata.connectors.http.internal.table.lookup;
 
 import java.io.Serializable;
-import java.util.Collections;
-import java.util.List;
 import java.util.Properties;
 
 import lombok.Builder;
 import lombok.Data;
 import lombok.RequiredArgsConstructor;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
 
 @Builder
 @Data
 @RequiredArgsConstructor
 public class HttpLookupConfig implements Serializable {
 
-    private final String url;
+    private final String lookupMethod;
 
-    @Builder.Default
-    private final List<String> arguments = Collections.emptyList();
+    private final String url;
 
     @Builder.Default
     private final boolean useAsync = false;
 
     @Builder.Default
     private final Properties properties = new Properties();
+
+    @Builder.Default
+    private final ReadableConfig readableConfig = new Configuration();
 }

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupConnectorOptions.java

@@ -35,4 +35,9 @@ public class HttpLookupConnectorOptions {
         ConfigOptions.key(SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER)
             .stringType()
             .noDefaultValue();
+
+    public static final ConfigOption<String> LOOKUP_REQUEST_FORMAT =
+        ConfigOptions.key("lookup-request.format")
+            .stringType()
+            .defaultValue("json");
 }