Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950649466
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -237,4 +419,48 @@ public DynamicTableSource copy() {
public String asSummaryString() {
return "Iceberg table source";
}
+
+ private Table loadTable() {
+ try (TableLoader tableLoader = loader.clone()) {
+ tableLoader.open();
+ return tableLoader.loadTable();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private boolean filtersSelectWholePartitions(Table table) {
+ if (filters == null || filters.isEmpty()) {
+ return true;
+ }
+
+ for (Expression filter : filters) {
+ if (!ExpressionUtil.selectsPartitions(filter, table, caseSensitive)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private Expression filterExpression() {
+ if (filters == null) {
+ return Expressions.alwaysTrue();
+ }
+
+ return filters.stream().reduce(Expressions.alwaysTrue(), Expressions::and);
+ }
+
+ private DataStream<RowData>
createAggregateDataStream(StreamExecutionEnvironment execEnv) {
+ RowData row =
+ new
StructRowData(pushedAggregate.resultType()).setStruct(pushedAggregate.result());
Review Comment:
You are right, build the row as a GenericRowData through
RowDataUtil.convertConstant.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
