@Override
        public <U> RelNode visit(TableSourceQueryOperation<U> tableSourceOperation) {
            TableSource<?> tableSource = tableSourceOperation.getTableSource();
            boolean isBatch;
            if (tableSource instanceof LookupableTableSource) {
                isBatch = tableSourceOperation.isBatch();
            } else if (tableSource instanceof StreamTableSource) {
                isBatch = ((StreamTableSource<?>) tableSource).isBounded();
            } else {
                throw new TableException(String.format("%s is not supported.", tableSource.getClass().getSimpleName()));
            }

            FlinkStatistic statistic;
            ObjectIdentifier tableIdentifier;
            if (tableSourceOperation instanceof RichTableSourceQueryOperation &&
                    ((RichTableSourceQueryOperation<U>) tableSourceOperation).getIdentifier() != null) {
                tableIdentifier = ((RichTableSourceQueryOperation<U>) tableSourceOperation).getIdentifier();
                statistic = ((RichTableSourceQueryOperation<U>) tableSourceOperation).getStatistic();
            } else {
                statistic = FlinkStatistic.UNKNOWN();
                // TableSourceScan requires a unique name of a Table for computing a digest.
                // We are using the identity hash of the TableSource object.
                String refId = "Unregistered_TableSource_" + System.identityHashCode(tableSource);
                CatalogManager catalogManager = relBuilder.getCluster().getPlanner().getContext()
                        .unwrap(FlinkContext.class).getCatalogManager();
                tableIdentifier = catalogManager.qualifyIdentifier(UnresolvedIdentifier.of(refId));
            }

            RelDataType rowType = TableSourceUtil.getSourceRowType(relBuilder.getTypeFactory(),
                    tableSourceOperation.getTableSchema(),
                    scala.Option.apply(tableSource),
                    !isBatch);
            TableSourceTable<?> tableSourceTable = new TableSourceTable<>(
                    relBuilder.getRelOptSchema(),
                    tableIdentifier,
                    rowType,
                    statistic,
                    tableSource,
                    !isBatch,
                    ConnectorCatalogTable.source(tableSource, isBatch));
            return LogicalTableScan.create(relBuilder.getCluster(), tableSourceTable);
        }
Get the {@link AggCall} correspond to the aggregate or table aggregate expression.
core/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java