Table API & SQL

/**
 * Main interface for Table API programs
 */
interface TableEnvironment {
    /**
     * Create table environment
     * @return Table environment instance
     */
    static TableEnvironment create();
    
    /**
     * Execute SQL query
     * @param query SQL query string
     * @return Table result
     */
    Table sqlQuery(String query);
    
    /**
     * Execute SQL statement
     * @param statement SQL statement
     * @return Table result
     */
    TableResult executeSql(String statement);
    
    /**
     * Get table from identifier
     * @param path Table path
     * @return Table
     */
    Table from(String path);
    
    /**
     * Create temporary view from table
     * @param path View path
     * @param table Table to create view from
     */
    void createTemporaryView(String path, Table table);
    
    /**
     * Register table
     * @param name Table name
     * @param table Table to register
     */
    void registerTable(String name, Table table);
    
    /**
     * Register function
     * @param name Function name
     * @param function User-defined function
     */
    void registerFunction(String name, UserDefinedFunction function);
}

/**
 * Stream table environment bridging Table and DataStream APIs
 */
interface StreamTableEnvironment extends TableEnvironment {
    /**
     * Create stream table environment from stream execution environment
     * @param executionEnvironment Stream execution environment
     * @return Stream table environment
     */
    static StreamTableEnvironment create(StreamExecutionEnvironment executionEnvironment);
    
    /**
     * Create stream table environment with settings
     * @param executionEnvironment Stream execution environment
     * @param settings Environment settings
     * @return Stream table environment
     */
    static StreamTableEnvironment create(
        StreamExecutionEnvironment executionEnvironment,
        EnvironmentSettings settings);
    
    /**
     * Convert table to append data stream
     * @param table Table to convert
     * @param clazz Target type class
     * @param <T> Target type
     * @return Data stream in append mode
     */
    <T> DataStream<T> toAppendStream(Table table, Class<T> clazz);
    
    /**
     * Convert table to retract data stream
     * @param table Table to convert
     * @param clazz Target type class
     * @param <T> Target type
     * @return Data stream with retract records
     */
    <T> DataStream<Tuple2<Boolean, T>> toRetractStream(Table table, Class<T> clazz);
    
    /**
     * Create table from data stream
     * @param dataStream Data stream
     * @param fields Field expressions
     * @param <T> Stream element type
     * @return Table
     */
    <T> Table fromDataStream(DataStream<T> dataStream, Expression... fields);
    
    /**
     * Register data stream as table
     * @param name Table name
     * @param dataStream Data stream
     * @param fields Field expressions
     * @param <T> Stream element type
     */
    <T> void registerDataStream(String name, DataStream<T> dataStream, Expression... fields);
}

/**
 * Environment settings for configuring table environments
 */
class EnvironmentSettings {
    /**
     * Create new builder for environment settings
     * @return Settings builder
     */
    public static Builder newInstance();
    
    /**
     * Create settings for streaming mode
     * @return Environment settings for streaming
     */
    public static EnvironmentSettings inStreamingMode();
    
    /**
     * Create settings for batch mode
     * @return Environment settings for batch
     */
    public static EnvironmentSettings inBatchMode();
    
    /**
     * Builder for environment settings
     */
    public static class Builder {
        /**
         * Use streaming planner
         * @return Builder
         */
        public Builder useStreamingPlanner();
        
        /**
         * Use batch planner
         * @return Builder
         */
        public Builder useBatchPlanner();
        
        /**
         * Use blink planner
         * @return Builder
         */
        public Builder useBlinkPlanner();
        
        /**
         * Use old planner
         * @return Builder
         */
        public Builder useOldPlanner();
        
        /**
         * Build environment settings
         * @return Environment settings
         */
        public EnvironmentSettings build();
    }
}

/**
 * Table interface for relational operations
 */
interface Table {
    /**
     * Select columns
     * @param fields Column expressions
     * @return Table with selected columns
     */
    Table select(Expression... fields);
    
    /**
     * Filter rows
     * @param predicate Filter predicate
     * @return Filtered table
     */
    Table where(Expression predicate);
    
    /**
     * Group by columns
     * @param fields Grouping columns
     * @return Grouped table
     */
    GroupedTable groupBy(Expression... fields);
    
    /**
     * Order by columns
     * @param fields Ordering expressions
     * @return Ordered table
     */
    Table orderBy(Expression... fields);
    
    /**
     * Limit number of rows
     * @param fetch Number of rows to fetch
     * @return Limited table
     */
    Table limit(int fetch);
    
    /**
     * Limit with offset
     * @param offset Number of rows to skip
     * @param fetch Number of rows to fetch
     * @return Limited table
     */
    Table limit(int offset, int fetch);
    
    /**
     * Join with another table
     * @param right Right table
     * @param joinPredicate Join condition
     * @return Joined table
     */
    Table join(Table right, Expression joinPredicate);
    
    /**
     * Left outer join
     * @param right Right table
     * @param joinPredicate Join condition
     * @return Joined table
     */
    Table leftOuterJoin(Table right, Expression joinPredicate);
    
    /**
     * Execute table and return result
     * @return Table result
     */
    TableResult execute();
    
    /**
     * Get table schema
     * @return Table schema
     */
    TableSchema getSchema();
}

/**
 * Grouped table operations
 */
interface GroupedTable {
    /**
     * Select with aggregations
     * @param fields Selection expressions with aggregations
     * @return Aggregated table
     */
    Table select(Expression... fields);
    
    /**
     * Aggregate using aggregate function
     * @param aggregateFunction Aggregate function
     * @return Aggregated table
     */
    AggregatedTable aggregate(Expression aggregateFunction);
}

/**
 * Aggregated table operations
 */
interface AggregatedTable {
    /**
     * Select from aggregated results
     * @param fields Selection expressions
     * @return Selected table
     */
    Table select(Expression... fields);
}

/**
 * Base class for user-defined functions
 */
abstract class UserDefinedFunction {
    /**
     * Get function kind
     * @return Function kind
     */
    public abstract FunctionKind getKind();
    
    /**
     * Get type inference
     * @return Type inference
     */
    public TypeInference getTypeInference(DataTypeFactory typeFactory);
}

/**
 * Scalar function base class
 */
abstract class ScalarFunction extends UserDefinedFunction {
    /**
     * Evaluation method - implement with desired signature
     * @param params Function parameters
     * @return Function result
     */
    public Object eval(Object... params);
    
    @Override
    public FunctionKind getKind() {
        return FunctionKind.SCALAR;
    }
}

/**
 * Table function base class
 * @param <T> Output row type
 */
abstract class TableFunction<T> extends UserDefinedFunction {
    /**
     * Emit row to collector
     * @param row Row to emit
     */
    protected void collect(T row);
    
    /**
     * Evaluation method - implement with desired signature
     * @param params Function parameters
     */
    public void eval(Object... params);
    
    @Override
    public FunctionKind getKind() {
        return FunctionKind.TABLE;
    }
}

/**
 * Aggregate function base class
 * @param <T> Result type
 * @param <ACC> Accumulator type
 */
abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
    /**
     * Create accumulator
     * @return New accumulator
     */
    public abstract ACC createAccumulator();
    
    /**
     * Get result from accumulator
     * @param accumulator Accumulator
     * @return Result
     */
    public abstract T getValue(ACC accumulator);
    
    /**
     * Accumulate value
     * @param accumulator Accumulator
     * @param params Input values
     */
    public abstract void accumulate(ACC accumulator, Object... params);
    
    @Override
    public FunctionKind getKind() {
        return FunctionKind.AGGREGATE;
    }
}

/**
 * Base interface for table expressions
 */
interface Expression {
    /**
     * Get expression SQL representation
     * @return SQL string
     */
    String asSummaryString();
}

/**
 * Factory methods for table expressions
 */
class Expressions {
    /**
     * Create column reference
     * @param name Column name
     * @return Column expression
     */
    public static Expression $(String name);
    
    /**
     * Create literal value
     * @param value Literal value
     * @return Literal expression
     */
    public static Expression lit(Object value);
    
    /**
     * Create range expression
     * @param from Start column
     * @param to End column
     * @return Range expression
     */
    public static Expression range(String from, String to);
    
    /**
     * Create all columns expression
     * @return All columns expression
     */
    public static Expression $star();
    
    // Arithmetic operations
    public static Expression plus(Expression left, Expression right);
    public static Expression minus(Expression left, Expression right);
    public static Expression times(Expression left, Expression right);
    public static Expression div(Expression left, Expression right);
    
    // Comparison operations
    public static Expression isEqual(Expression left, Expression right);
    public static Expression isNotEqual(Expression left, Expression right);
    public static Expression isLess(Expression left, Expression right);
    public static Expression isGreater(Expression left, Expression right);
    
    // Logical operations
    public static Expression and(Expression left, Expression right);
    public static Expression or(Expression left, Expression right);
    public static Expression not(Expression expression);
    
    // String operations
    public static Expression like(Expression str, Expression pattern);
    public static Expression upper(Expression str);
    public static Expression lower(Expression str);
    
    // Aggregate functions
    public static Expression count(Expression expression);
    public static Expression sum(Expression expression);
    public static Expression avg(Expression expression);
    public static Expression max(Expression expression);
    public static Expression min(Expression expression);
}

/**
 * Result of table operations and queries
 */
interface TableResult {
    /**
     * Get result schema
     * @return Table schema
     */
    TableSchema getTableSchema();
    
    /**
     * Print result to console
     */
    void print();
    
    /**
     * Get job client for result
     * @return Optional job client
     */
    Optional<JobClient> getJobClient();
    
    /**
     * Collect results
     * @return Iterator over rows
     */
    CloseableIterator<Row> collect();
    
    /**
     * Get result kind
     * @return Result kind
     */
    ResultKind getResultKind();
}

/**
 * Result kind enumeration
 */
enum ResultKind {
    SUCCESS,
    SUCCESS_WITH_CONTENT
}

/**
 * Logical data type representation
 */
abstract class DataType {
    /**
     * Get logical type
     * @return Logical type
     */
    public abstract LogicalType getLogicalType();
    
    /**
     * Get conversion class
     * @return Conversion class
     */
    public abstract Class<?> getConversionClass();
    
    /**
     * Create nullable version
     * @return Nullable data type
     */
    public abstract DataType nullable();
    
    /**
     * Create non-nullable version
     * @return Non-nullable data type
     */
    public abstract DataType notNull();
}

/**
 * Logical type system
 */
abstract class LogicalType {
    /**
     * Check if type is nullable
     * @return true if nullable
     */
    public abstract boolean isNullable();
    
    /**
     * Get type root
     * @return Type root
     */
    public abstract LogicalTypeRoot getTypeRoot();
    
    /**
     * Copy type
     * @param isNullable Nullable setting
     * @return Copied type
     */
    public abstract LogicalType copy(boolean isNullable);
}

/**
 * Row type for structured data
 */
class RowType extends LogicalType {
    /**
     * Get field names
     * @return List of field names
     */
    public List<String> getFieldNames();
    
    /**
     * Get field types
     * @return List of field types
     */
    public List<LogicalType> getFieldTypes();
    
    /**
     * Get field by name
     * @param fieldName Field name
     * @return Optional field
     */
    public Optional<RowType.RowField> getField(String fieldName);
}