AHeise commented on code in PR #56:
URL:
https://github.com/apache/flink-connector-http/pull/56#discussion_r4218448785
##########
flink-connector-http/src/main/java/org/apache/flink/connector/http/table/lookup/HttpLookupTableSource.java:
##########
@@ -178,12 +178,15 @@ protected LookupRuntimeProvider getLookupRuntimeProvider(
@Override
public DynamicTableSource copy() {
- return new HttpLookupTableSource(
- physicalRowDataType,
- lookupConfig,
- decodingFormat,
- dynamicTableFactoryContext,
- cache);
+ final HttpLookupTableSource copy =
+ new HttpLookupTableSource(
+ physicalRowDataType,
+ lookupConfig,
+ decodingFormat,
+ dynamicTableFactoryContext,
+ cache);
+ copy.metadataKeys = metadataKeys;
+ return copy;
Review Comment:
Cheap enough, added to the copy commit: equality over physical type, lookup
config, format, cache and metadata keys, and the copy test now asserts
`copy().isEqualTo(source)`. The factory context stays out since it is
construction input without equality.
##########
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:
That block is moved unchanged. The skipped entries are ROW-typed join keys
(`o.row = c.row`, see `testLookupJoinOnRowType`), which the planner does allow;
key paths into nested fields never reach the source since `extractLookupRow`
only walks top-level key indices. Booleans go through the single-value path
like any other scalar.
--
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]
