org.apache.flink.table.factories

Interface DynamicTableFactory.Context

All Known Implementing Classes:

FactoryUtil.DefaultDynamicTableContext

Enclosing interface:

DynamicTableFactory

@PublicEvolving
public static interface DynamicTableFactory.Context

Provides catalog and session information describing the dynamic table to be accessed.

Method Summary

Modifier and Type	Method and Description
ResolvedCatalogTable	getCatalogTable() Returns the resolved table information received from the Catalog or persisted plan.
ClassLoader	getClassLoader() Returns the class loader of the current session.
ReadableConfig	getConfiguration() Gives read-only access to the configuration of the current session.
default Map<String,String>	getEnrichmentOptions() Returns a map of options that can enrich the options of the original getCatalogTable() during a plan restore.
ObjectIdentifier	getObjectIdentifier() Returns the identifier of the table in the Catalog.
default DataType	getPhysicalRowDataType() Returns the physical schema to use for encoding and decoding records.
default int[]	getPrimaryKeyIndexes() Returns the primary key indexes, if any, otherwise returns an empty array.
boolean	isTemporary() Whether the table is temporary.

Method Detail

getObjectIdentifier

ObjectIdentifier getObjectIdentifier()

Returns the identifier of the table in the Catalog.

This identifier describes the relationship between the table instance and the associated Catalog (if any). However, it doesn't uniquely identify this specific table configuration. The same table might be stored in different catalogs or, in case of anonymous tables, this identifier is auto-generated and non-deterministic. Because of that behaviour, we strongly suggest using this identifier only for printing and logging purposes, and rely on user input for uniquely identifying a "table instance".

For example, when implementing a Kafka source using consumer groups, the user should provide the consumer group id manually rather than using this identifier as the consumer group id, so the offset tracking remains stable even if this table is anonymous, or it's moved to another Catalog.

Note that for anonymous tables ObjectIdentifier.asSerializableString() will fail, so we suggest to use ObjectIdentifier.asSummaryString() for generating strings.

getCatalogTable

ResolvedCatalogTable getCatalogTable()

Returns the resolved table information received from the Catalog or persisted plan.

The ResolvedCatalogTable forwards the metadata from the catalog but offers a validated ResolvedSchema. The original metadata object is available via ResolvedCatalogTable.getOrigin().

In most cases, a factory is interested in the following characteristics:

 // get the physical data type to initialize the connector
 context.getCatalogTable().getResolvedSchema().toPhysicalRowDataType()

 // get primary key information if the connector supports upserts
 context.getCatalogTable().getResolvedSchema().getPrimaryKey()

 // get configuration options
 context.getCatalogTable().getOptions()
 

Other characteristics such as metadata columns or watermarks will be pushed down into the created DynamicTableSource or DynamicTableSink during planning depending on the implemented ability interfaces.

During a plan restore, usually the table information persisted in the plan is used to reconstruct the catalog table. If and only if table.plan.restore.catalog-objects is set to ALL, there might be information from both the plan and a catalog lookup which requires considering getEnrichmentOptions(). It enables to enrich the plan information with frequently changing options (e.g. connection information or timeouts).

getEnrichmentOptions

default Map<String,String> getEnrichmentOptions()

Returns a map of options that can enrich the options of the original getCatalogTable() during a plan restore.

If and only if table.plan.restore.catalog-objects is set to ALL, this method may return a non-empty Map of options retrieved from the Catalog.

Because only the DynamicTableFactory is able to decide which options are safe to be forwarded without affecting the original topology, enrichment options are exposed through this method. In general, it's highly recommended using the FactoryUtil#createTableFactoryHelper(DynamicTableFactory, Context) to merge the options and then get the result with FactoryUtil.TableFactoryHelper.getOptions(). The helper considers both DynamicTableFactory.forwardOptions() and FormatFactory.forwardOptions().

Since a restored topology is static, an implementer has to ensure that the declared options don't affect fundamental abilities. The planner might not react to changed abilities anymore.

See Also:

FactoryUtil.TableFactoryHelper

getConfiguration

ReadableConfig getConfiguration()

Gives read-only access to the configuration of the current session.

getClassLoader

ClassLoader getClassLoader()

Returns the class loader of the current session.

The class loader is in particular useful for discovering further (nested) factories.

isTemporary

boolean isTemporary()

Whether the table is temporary.

getPhysicalRowDataType

default DataType getPhysicalRowDataType()

Returns the physical schema to use for encoding and decoding records. The returned row data type contains only physical columns. It does not include computed or metadata columns. A factory can use the returned data type to configure the table connector, and can manipulate it using the DataType static methods:

 // Project some fields into a new data type
 DataType projectedDataType = Projection.of(projectedIndexes)
     .project(context.getPhysicalRowDataType());

 // Create key data type
 DataType keyDataType = Projection.of(context.getPrimaryKeyIndexes())
     .project(context.getPhysicalRowDataType());

 // Create a new data type filtering columns of the original data type
 DataType myOwnDataType = DataTypes.ROW(
      DataType.getFields(context.getPhysicalRowDataType())
          .stream()
          .filter(myFieldFilterPredicate)
          .toArray(DataTypes.Field[]::new))
 

Shortcut for getCatalogTable().getResolvedSchema().toPhysicalRowDataType().

See Also:

ResolvedSchema.toPhysicalRowDataType()

getPrimaryKeyIndexes

default int[] getPrimaryKeyIndexes()

Returns the primary key indexes, if any, otherwise returns an empty array. A factory can use it to compute the schema projection of the key fields with Projection.of(ctx.getPrimaryKeyIndexes()).project(dataType).

Shortcut for getCatalogTable().getResolvedSchema().getPrimaryKeyIndexes().

See Also:

ResolvedSchema.getPrimaryKeyIndexes()