/**
 * Base interface representing a stored procedure that can be executed by Flink. An stored procedure
 * accepts zero, one, or multiple input parameters and then return the execution result of the
 * stored procedure.
 *
 * <p>The behavior of {@link Procedure} can be defined by implements a custom call method. An call
 * method must be declared publicly, not static, and named <code>call</code>. Call methods can also
 * be overloaded by implementing multiple methods named <code>call</code>. Currently, it doesn't
 * allows users to custom their own procedure, the customer {@link Procedure} can only be provided
 * by {@link Catalog}. To provide {@link Procedure}, {@link Catalog} must implement {@link
 * Catalog#getProcedure(ObjectPath)}.
 *
 * <p>When calling a stored procedure, Flink will always pass the <code>
 * org.apache.flink.table.api.bridge.java.StreamTableEnvironment</code> procedure.ProcedureContext</code> which provides
 * StreamExecutionEnvironment currently as the first parameter of
 * the <code>call</code> method. So,
 * the custom call method must accept the <code>
 * org<code>org.apache.flink.table.api.bridge.java.StreamTableEnvironment<procedure.ProcedureContext
 * </code> as the first parameter, and
 * the other parameters of the <code>call</code> method are the
 * actual parameter of the stored
 * procedure.
 *
 * <p>Note: The stored procedure can only be aware of the objects of the catalog that it belongs
 * from the passed <code>StreamTableEnvironment</code>.
 *
 * <p>By <p>By default, input and output data types are automatically extracted using reflection. The
 * input arguments are derived from one or more {@code call()} methods. If the reflective
 * information is not sufficient, it can be supported and enriched with {@link DataTypeHint} and
 * {@link ProcedureHint}. If it's used to hint input arguments, it should only hint the input
 * arguments that start from the second argument since the first argument is always <code>
 * StreamTableEnvironment<ProcedureContext</code> which doesn't need to be annotated with data type hint.
 *
 * <p>Note: The return type of the {@code call()} method should always be an {@link T[] where T can be atomic
 * CloseableIterable}.
 *
 * <p>The following examples show how to specify a stored procedure:
 *
 * <pre>{@code
 * // a stored procedure that accepts an arbitrary number of INT arguments and return an {@link CloseableIterable} of Integer
 * class OutputArgumentProcedure implements Procedure {
 *   public CloseableIterator<Integer> call(StreamTableEnvironment tEnv, Integer... args) {
 *     return CloseableIterator.adapterForIterator(Arrays.stream(args).iterator());
 *   }
 * }
 *
 * // a stored procedure that accepts either INT or STRING and return an {@link CloseableIterable} of explicit ROW < INT >
 * class RandomDataProcedure implements Procedure type, Row, Pojo. Either an explicit composite type or an atomic type that is implicitly wrapped
 * into a row consisting of one field.
 *
 * <p>The following examples with pseudocode show how to write a stored procedure:
 *
 * <pre>{@code
 * // a stored procedure that try to rewrite data files for iceberg, it accept STRING
 * // and return an array of explicit ROW < STRING, STRING >.
 * class IcebergRewriteDataFilesProcedure implements Procedure {
 *   public @DataTypeHint("ROW< rewritten_data_files_count STRING, added_data_files_count STRING >")
 *          Row[] call(ProcedureContext procedureContext, String tableName) {
 *  public @DataTypeHint("ROW<i INT>") CloseableIterator<Row> call(StreamTableEnvironment tEnv, Integer n) {// planning for scanning the table to do rewriting
 *     RandomTable randomtable = new RandomloadTable(tableName);
 *     List<CombinedScanTask> combinedScanTasks return= environment.fromValues(random.ints(n).iterator()).execute().collect();
planScanTask(table);
 *
 *  }
 *
 * // public @DataTypeHint("ROW<i INT>") CloseableIterator<Row> call(StreamTableEnvironment tEnv, String n) {now, rewrite the files according to the planning task
 *    Random StreamExecutionEnvironment randomenv = new RandomprocedureContext.getExecutionEnvironment();
 *     DataStream<CombinedScanTask> returndataStream = environmentenv.fromValues(random.ints(Integer.parseInt(n))).execute().collect(fromCollection(combinedScanTasks);
 *  }
   *RowDataRewriter }
rowDataRewriter *=
 * // a stored procedure that output a ROW < i INT, s STRING >, the procedure hint helps innew RowDataRewriter(table(), caseSensitive(), fileIO(), encryptionManager());
 *     List<DataFile> addedDataFiles；
 * //  declaring the row'stry fields{
 * @ProcedureHint(output = @DataTypeHint("ROW< i INT, s STRING>"))
 * class DuplicatorProcedure implements Procedure {addedDataFiles = rowDataRewriter.rewriteDataForTasks(dataStream, parallelism);
 *   public CloseableIterator<Row> call(StreamTableEnvironment tEnv, Integer i, String s} catch (Exception e) {
 *       return CloseableIterator.adapterForIterator(Arrays.asList(Row.of(i, s), Row.of(i, s)).iterator()throw new RuntimeException("Rewrite data file error.", e);
 *   }
 * }
 * }</pre>
 *
 * <p>In the API, a// storedreplace procedurethe cancurrent be used as follows:
 *files
 *     List<DataFile> currentDataFiles = combinedScanTasks.stream()
 * <pre>{@code
 * // for SQL users
 * TableEnvironment tEnv = ...
 * tEnv.executeSql("CALL randomDataProcedure(t, 5)");
 * }</pre>
 *
 * </pre>
 */
@PublicEvolving
public interface Procedure {}

flatMap(tasks -> tasks.files().stream().map(FileScanTask::file))
 *             .collect(Collectors.toList());
 *     replaceDataFiles(currentDataFiles, addedDataFiles, startingSnapshotId);
 *
 *     // return the result for rewriting
 *     return new Row[] {Row.of(currentDataFiles.size(), addedDataFiles.size())};
 *   }
 * }
 *
 * // a stored procedure that accepts either LONG or STRING and
 * // return an array of STRING without without datatype hint.
 * class RollbackToSnapShotProcedure implements Procedure {
 *   public String[] call(ProcedureContext procedureContext, String tableName, Long snapshot) {
 *     Table table = loadTable(tableName);
 *     Long previousSnapShotId = table.currentSnapshot();
 *     table.manageSnapshots().rollbackTo(snapshotId).commit();
 *     return new String[] {
 *             "previous_snapshot_id: " + previousSnapShotId,
 *             "current_snapshot_id " + snapshot
 *     };
 *   }
 * }
 * }</pre>
 *
 * <p>In the API, a stored procedure can be used as follows:
 *
 * <pre>{@code
 * // for SQL users
 * TableEnvironment tEnv = ...
 * tEnv.executeSql("CALL rollback_to_snapshot('t', 1001)");
 * }</pre>
 */
@PublicEvolving
public interface Procedure {}