private RelNode convertToDataStreamScan(DataStreamQueryOperation<?> operation) {
            List<String> names;
            ObjectIdentifier identifier = operation.getIdentifier();
            if (identifier != null) {
                names = Arrays.asList(
                        identifier.getCatalogName(),
                        identifier.getDatabaseName(),
                        identifier.getObjectName());
            } else {
                String refId = String.format("Unregistered_DataStream_%s", operation.getDataStream().getId());
                names = Collections.singletonList(refId);
            }

            final RelDataType rowType = DataStreamTable$.MODULE$
                    .getRowType(relBuilder.getTypeFactory(),
                            operation.getDataStream(),
                            operation.getTableSchema().getFieldNames(),
                            operation.getFieldIndices(),
                            scala.Option.apply(operation.getFieldNullables()));
            DataStreamTable<?> dataStreamTable = new DataStreamTable<>(
                    relBuilder.getRelOptSchema(),
                    names,
                    rowType,
                    operation.getDataStream(),
                    operation.isProducesUpdates(),
                    operation.isAccRetract(),
                    operation.getFieldIndices(),
                    operation.getTableSchema().getFieldNames(),
                    operation.getStatistic(),
                    scala.Option.apply(operation.getFieldNullables()));
            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