JingsongLi commented on a change in pull request #13729:
URL: https://github.com/apache/flink/pull/13729#discussion_r517155968
##########
File path:
flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FilesystemLookupFunction.java
##########
@@ -51,125 +45,113 @@
import java.util.stream.IntStream;
/**
- * Lookup table function for filesystem connector tables.
+ * Lookup function for filesystem connector tables.
+ *
+ * <p>The hive connector and filesystem connector share read/write files code.
+ * Currently, only this function only used in hive connector.
*/
-public class FileSystemLookupFunction<T extends InputSplit> extends
TableFunction<RowData> {
-
- private static final long serialVersionUID = 1L;
+public class FilesystemLookupFunction<P> extends TableFunction<RowData> {
- private static final Logger LOG =
LoggerFactory.getLogger(FileSystemLookupFunction.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(FilesystemLookupFunction.class);
// the max number of retries before throwing exception, in case of
failure to load the table into cache
private static final int MAX_RETRIES = 3;
// interval between retries
private static final Duration RETRY_INTERVAL = Duration.ofSeconds(10);
- private final InputFormat<RowData, T> inputFormat;
- // names and types of the records returned by the input format
- private final String[] producedNames;
- private final DataType[] producedTypes;
+ private final PartitionFetcher<P> partitionFetcher;
+ private final PartitionReader<P, RowData> partitionReader;
+ private final int[] lookupCols;
+ private final RowData.FieldGetter[] lookupFieldGetters;
private final Duration cacheTTL;
+ private final TypeSerializer<RowData> serializer;
+ private final DataType[] fieldTypes;
+ private final String[] fieldNames;
- // indices of lookup columns in the record returned by input format
- private final int[] lookupCols;
- // use Row as key for the cache
- private transient Map<Row, List<RowData>> cache;
+ // cache for lookup data
+ private transient Map<RowData, List<RowData>> cache;
// timestamp when cache expires
private transient long nextLoadTime;
- // serializer to copy RowData
- private transient TypeSerializer<RowData> serializer;
- // converters to convert data from internal to external in order to
generate keys for the cache
- private final DataFormatConverter[] converters;
- public FileSystemLookupFunction(
- InputFormat<RowData, T> inputFormat,
+ public FilesystemLookupFunction(
+ PartitionFetcher<P> partitionFetcher,
+ PartitionReader<P, RowData> partitionReader,
+ DataType[] fieldTypes,
+ String[] fieldNames,
String[] lookupKeys,
Review comment:
Can you use `int[] lookupKeys`?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
