Table Integration

/**
 * Parquet format factory for file system connectors
 */
public class ParquetFileFormatFactory implements BulkReaderFormatFactory, BulkWriterFormatFactory {
    
    /**
     * Creates a bulk decoding format for reading Parquet files
     * @param context Dynamic table factory context
     * @param formatOptions Configuration options
     * @return BulkDecodingFormat for reading RowData from Parquet files
     */
    public BulkDecodingFormat<RowData> createDecodingFormat(
        DynamicTableFactory.Context context, 
        ReadableConfig formatOptions
    );
    
    /**
     * Creates an encoding format for writing Parquet files  
     * @param context Dynamic table factory context
     * @param formatOptions Configuration options
     * @return EncodingFormat for writing RowData to Parquet files
     */
    public EncodingFormat<BulkWriter.Factory<RowData>> createEncodingFormat(
        DynamicTableFactory.Context context, 
        ReadableConfig formatOptions
    );
    
    /**
     * Returns the format identifier for this factory
     * @return "parquet"
     */
    public String factoryIdentifier();
    
    /**
     * Required configuration options (none for Parquet)
     * @return Empty set
     */
    public Set<ConfigOption<?>> requiredOptions();
    
    /**
     * Optional configuration options 
     * @return Set of supported configuration options
     */
    public Set<ConfigOption<?>> optionalOptions();
}

/**
 * Format identifier constant
 */
public static final String IDENTIFIER = "parquet";

/**
 * Use UTC timezone or local timezone for timestamp conversion
 * Default: false (use local timezone)
 */
public static final ConfigOption<Boolean> UTC_TIMEZONE = 
    key("utc-timezone")
        .booleanType()
        .defaultValue(false);

/**
 * Time unit for storing Parquet timestamps
 * Values: "nanos", "micros", "millis"
 * Default: "micros"
 */
public static final ConfigOption<String> TIMESTAMP_TIME_UNIT = 
    key("timestamp.time.unit")
        .stringType()
        .defaultValue("micros");

/**
 * Write timestamps as int64/LogicalTypes instead of int96/OriginalTypes
 * Default: false (use int96 format)
 */
public static final ConfigOption<Boolean> WRITE_INT64_TIMESTAMP = 
    key("write.int64.timestamp")
        .booleanType()
        .defaultValue(false);

/**
 * Batch size for vectorized reading
 * Default: 2048 rows per batch
 */
public static final ConfigOption<Integer> BATCH_SIZE = 
    key("batch-size")
        .intType()
        .defaultValue(2048);

/**
 * Bulk decoding format for reading Parquet files with statistics support
 */
public static class ParquetBulkDecodingFormat 
    implements ProjectableDecodingFormat<BulkFormat<RowData, FileSourceSplit>>,
               BulkDecodingFormat<RowData>,
               FileBasedStatisticsReportableInputFormat {
    
    /**
     * Creates runtime decoder with projection support
     * @param sourceContext Table source context
     * @param producedDataType Output data type
     * @param projections Column projections 
     * @return BulkFormat for reading projected RowData
     */
    public BulkFormat<RowData, FileSourceSplit> createRuntimeDecoder(
        DynamicTableSource.Context sourceContext,
        DataType producedDataType,
        int[][] projections
    );
    
    /**
     * Returns supported changelog mode (insert-only)
     * @return ChangelogMode.insertOnly()
     */
    public ChangelogMode getChangelogMode();
    
    /**
     * Reports table statistics from Parquet file metadata
     * @param files List of Parquet files to analyze
     * @param producedDataType Expected output data type
     * @return TableStats with row counts and column statistics
     */
    public TableStats reportStatistics(List<Path> files, DataType producedDataType);
}

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

TableEnvironment tableEnv = TableEnvironment.create(settings);

// Create table with Parquet format
tableEnv.executeSql("""
    CREATE TABLE orders (
        order_id BIGINT,
        customer_id STRING,
        product_name STRING,
        quantity INT,
        price DECIMAL(10,2),
        order_date TIMESTAMP(3)
    ) WITH (
        'connector' = 'filesystem',
        'path' = '/data/orders',
        'format' = 'parquet'
    )
""");

// Table with custom Parquet settings
tableEnv.executeSql("""
    CREATE TABLE events (
        event_id BIGINT,
        timestamp_col TIMESTAMP(3),
        payload STRING
    ) WITH (
        'connector' = 'filesystem',
        'path' = '/data/events',
        'format' = 'parquet',
        'parquet.utc-timezone' = 'true',
        'parquet.timestamp.time.unit' = 'nanos',
        'parquet.write.int64.timestamp' = 'true',
        'parquet.batch-size' = '4096'
    )
""");

// Only read specific columns for better performance
Table result = tableEnv.sqlQuery("""
    SELECT order_id, customer_id, price 
    FROM orders 
    WHERE order_date >= TIMESTAMP '2023-01-01 00:00:00'
""");

// Partitioned Parquet table
tableEnv.executeSql("""
    CREATE TABLE sales_partitioned (
        transaction_id BIGINT,
        amount DECIMAL(10,2),
        product_category STRING,
        sale_date DATE
    ) PARTITIONED BY (sale_date) WITH (
        'connector' = 'filesystem',
        'path' = '/data/sales',
        'format' = 'parquet',
        'sink.partition-commit.policy.kind' = 'success-file'
    )
""");