private RelNode convertToDataStreamScan(
                DataStream<?> dataStream,
                int[] fieldIndices,
                TableSchema tableSchema,
                Optional<ObjectIdentifier> identifier) {
            List<String> names;
            if (identifier.isPresent()) {
                names = Arrays.asList(
                        identifier.get().getCatalogName(),
                        identifier.get().getDatabaseName(),
                        identifier.get().getObjectName());
            } else {
                String refId = String.format("Unregistered_DataStream_%s", dataStream.getId());
                names = Collections.singletonList(refId);
            }
            final RelDataType rowType = DataStreamTable$.MODULE$
                    .getRowType(relBuilder.getTypeFactory(),
                            dataStream,
                            tableSchema.getFieldNames(),
                            fieldIndices,
                            scala.Option.empty());
            DataStreamTable<?> dataStreamTable = new DataStreamTable<>(
                    relBuilder.getRelOptSchema(),
                    names,
                    rowType,
                    dataStream,
                    producesUpdates,
                    false,
                    fieldIndices,
                    tableSchema.getFieldNames(),
                    FlinkStatistic.UNKNOWN(),
                    scala.Option.empty());
            return LogicalTableScan.create(relBuilder.getCluster(), dataStreamTable);
        }
Get the {@link AggCall} correspond to the aggregate or table aggregate expression.
core/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java