Table Environment and Setup

/**
 * Creates a table environment based on the provided settings
 * @param settings Configuration settings for the table environment
 * @return Configured TableEnvironment instance
 */
public static TableEnvironment create(EnvironmentSettings settings);

// Streaming mode environment
EnvironmentSettings streamSettings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()
    .build();
TableEnvironment streamEnv = TableEnvironment.create(streamSettings);

// Batch mode environment  
EnvironmentSettings batchSettings = EnvironmentSettings
    .newInstance()
    .inBatchMode()
    .build();
TableEnvironment batchEnv = TableEnvironment.create(batchSettings);

/**
 * Gets a table from the catalog by path
 * @param path Table path in format [[catalog.]database.]table
 * @return Table instance for the specified path
 */
public Table from(String path);

/**
 * Creates a table in the catalog with the specified descriptor
 * @param path Table path where to create the table
 * @param descriptor Table descriptor defining schema and properties
 */
public void createTable(String path, TableDescriptor descriptor);

/**
 * Creates a temporary view from a Table object
 * @param path View name/path
 * @param view Table object to create view from
 */
public void createTemporaryView(String path, Table view);

/**
 * Creates a temporary table from a descriptor
 * @param path Table path
 * @param descriptor Table descriptor
 */
public void createTemporaryTable(String path, TableDescriptor descriptor);

// Access existing table
Table orders = tableEnv.from("catalog1.database1.orders");
Table customers = tableEnv.from("customers"); // uses current catalog/database

// Create table with descriptor
TableDescriptor descriptor = TableDescriptor
    .forConnector("filesystem")
    .schema(Schema.newBuilder()
        .column("id", DataTypes.BIGINT())
        .column("name", DataTypes.STRING())
        .column("price", DataTypes.DECIMAL(10, 2))
        .build())
    .option("path", "/path/to/data")
    .option("format", "csv")
    .build();

tableEnv.createTable("my_table", descriptor);

// Create temporary view
Table filteredOrders = orders.filter($("amount").isGreater(100));
tableEnv.createTemporaryView("large_orders", filteredOrders);

/**
 * Executes a SQL query and returns the result as a Table
 * @param query SQL SELECT query
 * @return Table containing query results
 */
public Table sqlQuery(String query);

/**
 * Executes a SQL statement (DDL, DML, or query)
 * @param statement SQL statement to execute
 * @return TableResult containing execution results
 */
public TableResult executeSql(String statement);

// SQL query
Table result = tableEnv.sqlQuery(
    "SELECT customer_id, SUM(amount) as total " +
    "FROM orders " +
    "WHERE order_date >= '2023-01-01' " +
    "GROUP BY customer_id"
);

// SQL DDL statement
tableEnv.executeSql(
    "CREATE TABLE user_behavior (" +
    "  user_id BIGINT," +
    "  item_id BIGINT," +
    "  category_id BIGINT," +
    "  behavior STRING," +
    "  ts TIMESTAMP(3)" +
    ") WITH (" +
    "  'connector' = 'kafka'," +
    "  'topic' = 'user_behavior'," +
    "  'properties.bootstrap.servers' = 'localhost:9092'" +
    ")"
);

// SQL DML statement
tableEnv.executeSql(
    "INSERT INTO result_table " +
    "SELECT user_id, COUNT(*) as behavior_cnt " +
    "FROM user_behavior " +
    "GROUP BY user_id"
);

/**
 * Register a catalog with the given name
 * @param catalogName Name to register the catalog under
 * @param catalog Catalog instance to register
 */
public void registerCatalog(String catalogName, Catalog catalog);

/**
 * Set the current catalog for resolving unqualified table references
 * @param catalogName Name of catalog to use as current
 */
public void useCatalog(String catalogName);

/**
 * Set the current database within the current catalog
 * @param databaseName Name of database to use as current
 */
public void useDatabase(String databaseName);

/**
 * Get the current catalog name
 * @return Name of the current catalog
 */
public String getCurrentCatalog();

/**
 * Get the current database name
 * @return Name of the current database
 */
public String getCurrentDatabase();

/**
 * List all registered catalog names
 * @return Array of catalog names
 */
public String[] listCatalogs();

/**
 * List all databases in the current catalog
 * @return Array of database names
 */
public String[] listDatabases();

/**
 * List all tables in the current database
 * @return Array of table names
 */
public String[] listTables();

/**
 * Register a scalar function under the given name
 * @param name Function name to register under
 * @param function ScalarFunction instance
 */
public void registerFunction(String name, ScalarFunction function);

/**
 * Create a temporary system function from a class name
 * @param name Function name
 * @param functionClass Function class name
 */
public void createTemporarySystemFunction(String name, String functionClass);

/**
 * Create a temporary function from a function instance
 * @param path Function path/name
 * @param function Function instance
 */
public void createTemporaryFunction(String path, UserDefinedFunction function);

/**
 * Get the table configuration for this environment
 * @return TableConfig instance
 */
public TableConfig getConfig();

/**
 * Create a StatementSet for executing multiple statements together
 * @return StatementSet instance
 */
public StatementSet createStatementSet();

/**
 * Creates a new EnvironmentSettings builder
 * @return Builder for configuring environment settings
 */
public static EnvironmentSettings.Builder newInstance();

public static class Builder {
    /**
     * Sets the planner to Blink planner (default and only option in newer versions)
     * @return Builder instance for chaining
     */
    public Builder useBlinkPlanner();
    
    /**
     * Configures environment for streaming mode execution
     * @return Builder instance for chaining
     */
    public Builder inStreamingMode();
    
    /**
     * Configures environment for batch mode execution
     * @return Builder instance for chaining
     */
    public Builder inBatchMode();
    
    /**
     * Sets a custom class loader for the environment
     * @param classLoader Custom class loader
     * @return Builder instance for chaining
     */
    public Builder withClassLoader(ClassLoader classLoader);
    
    /**
     * Builds the EnvironmentSettings with configured options
     * @return EnvironmentSettings instance
     */
    public EnvironmentSettings build();
}

// Default streaming environment
EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()
    .build();

// Batch environment with custom class loader
ClassLoader customClassLoader = // ... custom class loader
EnvironmentSettings batchSettings = EnvironmentSettings
    .newInstance()
    .inBatchMode()
    .withClassLoader(customClassLoader)
    .build();

public final class TableConfig implements WritableConfig, ReadableConfig {
    /**
     * Get the underlying configuration object
     * @return Configuration instance
     */
    public Configuration getConfiguration();
    
    /**
     * Set the SQL dialect for parsing SQL statements
     * @param dialect SQL dialect to use
     */
    public void setSqlDialect(SqlDialect dialect);
    
    /**
     * Set the parallelism for table operations
     * @param parallelism Parallelism level
     */
    public void setParallelism(int parallelism);
    
    /**
     * Get the current parallelism setting
     * @return Current parallelism level
     */
    public int getParallelism();
}

public enum SqlDialect {
    /** Default Flink SQL dialect */
    DEFAULT,
    /** Hive-compatible SQL dialect */
    HIVE
}