Factory Classes and Service Providers

Factory classes provide the entry points for creating planner components through the Service Provider Interface (SPI). These factories are automatically discovered and instantiated by Flink's factory system.

Package Information

import org.apache.flink.table.planner.delegation.DefaultPlannerFactory;
import org.apache.flink.table.planner.delegation.DefaultParserFactory;
import org.apache.flink.table.planner.delegation.DefaultExecutorFactory;
import org.apache.flink.table.factories.PlannerFactory;
import org.apache.flink.table.factories.ParserFactory;
import org.apache.flink.table.factories.ExecutorFactory;

Capabilities

Default Planner Factory

Creates the default Planner implementation based on runtime mode (streaming or batch).

public final class DefaultPlannerFactory implements PlannerFactory {
    public String factoryIdentifier();
    public Planner create(Context context);
    public Set<ConfigOption<?>> requiredOptions();
    public Set<ConfigOption<?>> optionalOptions();
}

The DefaultPlannerFactory is the primary factory for creating planner instances. It automatically determines whether to create a StreamPlanner or BatchPlanner based on the runtime mode specified in the configuration.

Key Methods:

• factoryIdentifier(): Returns PlannerFactory.DEFAULT_IDENTIFIER to identify this factory
• create(Context context): Creates either StreamPlanner (streaming mode) or BatchPlanner (batch mode)
• requiredOptions(): Returns empty set - no required configuration options
• optionalOptions(): Returns empty set - no optional configuration options

Usage Example:

import org.apache.flink.table.planner.delegation.DefaultPlannerFactory;
import org.apache.flink.table.factories.PlannerFactory;

// Factory is typically discovered automatically via SPI
PlannerFactory factory = new DefaultPlannerFactory();
String identifier = factory.factoryIdentifier(); // Returns "default"

// Create planner with context
Planner planner = factory.create(context);

Default Parser Factory

Creates ParserImpl instances for SQL parsing using Apache Calcite.

public class DefaultParserFactory implements ParserFactory {
    public String factoryIdentifier();
    public Parser create(Context context);
    public Set<ConfigOption<?>> requiredOptions();
    public Set<ConfigOption<?>> optionalOptions();
}

The DefaultParserFactory creates instances of ParserImpl which handle SQL statement parsing, identifier parsing, and SQL expression parsing.

Key Methods:

• factoryIdentifier(): Returns the default SQL dialect name in lowercase
• create(Context context): Creates a new ParserImpl instance
• requiredOptions(): Returns empty set - no required configuration options
• optionalOptions(): Returns empty set - no optional configuration options

Usage Example:

import org.apache.flink.table.planner.delegation.DefaultParserFactory;
import org.apache.flink.table.factories.ParserFactory;

// Create parser factory
ParserFactory parserFactory = new DefaultParserFactory();
String identifier = parserFactory.factoryIdentifier(); // Returns "default"

// Create parser with context
Parser parser = parserFactory.create(context);

// Parse SQL statements
List<Operation> operations = parser.parse("SELECT * FROM my_table");

Default Executor Factory

Creates DefaultExecutor instances for executing table programs.

public final class DefaultExecutorFactory implements ExecutorFactory {
    public Executor create(Context context);
    public String factoryIdentifier();
    public Set<ConfigOption<?>> requiredOptions();
    public Set<ConfigOption<?>> optionalOptions();
}

The DefaultExecutorFactory creates instances of DefaultExecutor which handle the execution of table programs converted from the planner.

Key Methods:

• factoryIdentifier(): Returns the factory identifier for this executor
• create(Context context): Creates a new DefaultExecutor instance
• requiredOptions(): Returns empty set - no required configuration options
• optionalOptions(): Returns empty set - no optional configuration options

Usage Example:

import org.apache.flink.table.planner.delegation.DefaultExecutorFactory;
import org.apache.flink.table.factories.ExecutorFactory;

// Create executor factory
ExecutorFactory executorFactory = new DefaultExecutorFactory();
String identifier = executorFactory.factoryIdentifier();

// Create executor with context
Executor executor = executorFactory.create(context);

// Execute table programs
JobExecutionResult result = executor.execute(transformations);

Service Provider Interface (SPI) Registration

These factories are automatically registered through Java's Service Provider Interface mechanism. The registration is defined in:

META-INF/services/org.apache.flink.table.factories.Factory

This file contains the fully qualified class names:

• org.apache.flink.table.planner.delegation.DefaultPlannerFactory
• org.apache.flink.table.planner.delegation.DefaultParserFactory
• org.apache.flink.table.planner.delegation.DefaultExecutorFactory

Factory Context

All factories receive a Context object that provides access to:

public interface Context {
    Configuration getConfiguration();
    ClassLoader getClassLoader();
    TableEnvironment getTableEnvironment();
}

The context allows factories to:

• Access configuration settings for customizing behavior
• Use the appropriate class loader for loading resources
• Access the table environment for integration with existing components

Integration with Table Environment

The factories are typically used indirectly when creating a TableEnvironment:

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

// The planner factory is used internally when creating the environment
TableEnvironment tableEnv = TableEnvironment.create(
    EnvironmentSettings.newInstance()
        .useBlinkPlanner() // Uses DefaultPlannerFactory
        .build()
);

Factory Discovery Process

Flink's factory system uses the following discovery process:

1. SPI Discovery: Scan the classpath for META-INF/services/org.apache.flink.table.factories.Factory files
2. Factory Instantiation: Create instances of all discovered factory classes
3. Identifier Matching: Match factory identifiers with requested components
4. Context Creation: Create appropriate context objects with configuration
5. Component Creation: Call factory methods to create the actual components

This design enables pluggable architectures where different planner implementations can be provided by different modules or third-party libraries.