davidradl commented on code in PR #56:
URL:
https://github.com/apache/flink-connector-http/pull/56#discussion_r4217387079
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpTableLookupFunction.java:
##########
@@ -107,82 +106,68 @@ public Collection<RowData> lookup(RowData keyRow) {
HttpRowDataWrapper httpRowDataWrapper = client.pull(keyRow);
Collection<RowData> httpCollector = httpRowDataWrapper.getData();
- int physicalArity = -1;
-
- GenericRowData producedRow = null;
- // grab the actual data if there is any from the response and populate
the producedRow with
- // it
- if (!httpCollector.isEmpty()) {
- GenericRowData physicalRow = (GenericRowData)
httpCollector.iterator().next();
- physicalArity = physicalRow.getArity();
- producedRow =
- new GenericRowData(physicalRow.getRowKind(), physicalArity
+ metadataArity);
- // Build a map of lookup table field names to their typed values
from keyRow
- // Only process top-level single-value join keys
- Map<String, Object> joinKeyValues = new HashMap<>();
- for (LookupSchemaEntry<RowData> entry :
lookupRow.getLookupEntries()) {
- // Only handle top-level single value entries (not nested
RowTypeLookupSchemaEntry)
- if (entry instanceof RowDataSingleValueLookupSchemaEntry) {
- RowDataSingleValueLookupSchemaEntry singleEntry =
- (RowDataSingleValueLookupSchemaEntry) entry;
- try {
- // Get the typed value directly from keyRow using
fieldGetter
- Object typedValue =
singleEntry.fieldGetter.getFieldOrNull(keyRow);
- if (typedValue != null) {
- // Get the lookup table field name from LookupArg
- List<LookupArg> lookupArgs =
entry.convertToLookupArg(keyRow);
- for (LookupArg lookupArg : lookupArgs) {
- // Map lookup table field name to typed value
- joinKeyValues.put(lookupArg.getArgName(),
typedValue);
- }
+ if (httpCollector.isEmpty() && producedDataType == null) {
+ // If producedDataType is null and we have no data, return the
same way as ignore.
+ return Collections.emptyList();
+ }
+ final GenericRowData physicalRow =
+ httpCollector.isEmpty()
+ ? new GenericRowData(
+ RowKind.INSERT,
+ producedDataType.getChildren().size() -
metadataArity)
+ : (GenericRowData) httpCollector.iterator().next();
+ final int physicalArity = physicalRow.getArity();
+ GenericRowData producedRow =
+ new GenericRowData(physicalRow.getRowKind(), physicalArity +
metadataArity);
+ // Build a map of lookup table field names to their typed values from
keyRow
+ // Only process top-level single-value join keys
Review Comment:
Is it possible to get non top-level join keys here? I had thought the
planner would not allow this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
