Apache Paimon：存储结构、写入及其源码分析-EW帮帮网

Apache Paimon (此前称为 Flink Table Store)是一种流式数据湖存储技术，采用 LSM（Log-Structured Merge-tree）树结构来存储数据，支持高吞吐、低延迟的数据摄入和实时查询，尤其适用于流式和批量统一的场景。

1. 创建表 (CREATE TABLE)

当执行 CREATE TABLE 语句时（例如通过 Flink SQL 或 Spark SQL），Paimon 会执行以下操作：

CREATE TABLE IF NOT EXISTS user_behavior (
  user_id BIGINT,
  item_id BIGINT,
  behavior STRING,
  ts TIMESTAMP(3),
  PRIMARY KEY (ts, user_id, item_id) NOT ENFORCED
) PARTITIONED BY (ts)
WITH (
  'bucket-key' = 'user_id',
  'bucket' = '4',
  'snapshot.time-retained' = '1 h'
)

每个分区进一步分桶（Bucket），桶是读写的最小单元。默认使用列式存储（如 Parquet、ORC），数据文件按主键排序。

SQL 解析与 Catalog: 计算引擎（Flink/Spark）解析 SQL DDL 语句。请求通过 Paimon 的 Catalog 接口传递给 Paimon。Catalog 负责管理数据库和表的元数据。
Schema 管理:
- SchemaManager (源码路径: paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java) 负责处理表的 Schema 信息。它会接收用户定义的列、类型、主键、分区键、表属性等。
- Schema 信息会被持久化。在文件系统类型的 Catalog 中，Schema 会以 JSON 文件的形式存储在表的基础路径下的 schema 目录中，例如 warehouse/your_db/your_table/schema/schema-0。"0" 代表 Schema 的版本号。
表对象实例化:
- FileStoreTableFactory (源码路径: paimon-core/src/main/java/org/apache/paimon/table/FileStoreTableFactory.java) 会根据表的配置（例如是否有主键、分区信息等）创建相应的表实例。
  - 如果定义了主键，通常会创建 PrimaryKeyFileStoreTable (源码路径: paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java)。
  - 如果没有主键（追加表），则会创建 AppendOnlyFileStoreTable (源码路径: paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java)。
- 此时，磁盘上会创建表的基础目录结构，主要包含 schema 目录。snapshot 和 manifest 目录此时可能为空或不存在，直到第一次数据提交。

2. Paimon 文件组织 (File Layout)

Paimon 的文件以分层的方式组织，核心元数据包括快照 (Snapshot)、清单列表 (Manifest List) 和清单文件 (Manifest File)，它们共同指向实际的数据文件 (Data File)。

可以参考以下理解其文件布局：

/tmp/paimon/default.db/T/
├── schema/          # 存储表结构（schema-0）
├── snapshot/        # 快照目录（初始为空）
├── manifest/        # 清单文件目录（初始为空）
└── ts={partition}/  # 分区目录（按分区键动态生成）
    ├── bucket-0/               # 桶0目录
    │   ├── data-xxx.parquet
    │   └── ...
    ├── bucket-1/               # 桶1目录
    │   ├── data-yyy.parquet
        │   └── ...
    ├── bucket-2/
    └── bucket-3/

表目录 (Table Directory): 所有与该表相关的文件都存储在这个目录下，例如 file:///tmp/paimon/default.db/T。
- schema/: 存储表的 Schema 文件 (如 schema-0, schema-1, ...)。
- snapshot/: 存储快照文件 (如 snapshot-1, snapshot-2, ...)。
- manifest/: 存储清单列表文件和清单文件 (如 manifest-list-xxx, manifest-file-yyy, ...)。
- 分区目录 (如 dt=2023-01-01/): 如果表是分区的，数据文件会存储在相应的分区目录下。
- 桶目录 (如 bucket-0/): 在每个分区内（或表级，如果未分区），数据按桶组织。
Snapshot (快照):
- 每个成功的 commit 操作都会生成一个新的快照。快照代表了表在某个特定时间点的完整、一致的状态。
- 快照文件是一个小的 JSON 文件，存储在 snapshot/ 目录下。它包含了该快照的 ID、对应的 Schema ID、指向的 Manifest List 文件名、提交用户、提交时间等元信息。
- 例如 snapshot-1 的内容可能类似：
```
{
  "version" : 3,
  "id" : 1,
  "schemaId" : 0,
  "baseManifestList" : "manifest-list-uuid1-0",
  "deltaManifestList" : "manifest-list-uuid1-1",
  "commitUser" : "user-uuid",
  "commitIdentifier" : 1234567890,
  "commitKind" : "APPEND",
  "timeMillis" : 1672531200000
}
```
Manifest List (清单列表):
- 每个快照文件会指向一个 Manifest List 文件。
- Manifest List 文件本身也是一个元数据文件，它记录了一个或多个 Manifest 文件的列表。这些 Manifest 文件共同构成了该快照的数据视图。
- 存储在 manifest/ 目录下。
Manifest File (清单文件):
- 每个 Manifest 文件包含了一组数据文件 (Data File) 或 Changelog 文件的元数据。
- 这些元数据包括：数据文件名、文件所属的分区和桶、文件在 LSM 树中的层级 (level)、文件中记录的数量、文件中 key 的统计信息 (min/max key，用于数据跳过)、文件大小等。
- 存储在 manifest/ 目录下。
Data File (数据文件):
- 实际存储表记录的文件。Paimon 支持多种列式存储格式，如 Parquet (默认)、ORC、Avro。
- 数据文件根据表的分区键和桶键进行组织。例如，对于按 dt 分区、bucket 为 10 的表，一个数据文件路径可能为 your_warehouse/your_db/your_table/dt=2024-01-01/bucket-5/data-file-uuid.parquet。
Index File (索引文件): (可选)
- Paimon 支持为数据文件创建索引（例如布隆过滤器、MinMax 索引），这些索引信息可以存储在单独的索引文件中，或直接内嵌在 Manifest 文件中（如果索引较小）。

Manifest 文件的组织与存储格式

Paimon 的 Manifest 文件以及 Manifest List 文件是二进制存储的。它们通常使用 Apache Avro 格式进行序列化和存储。Avro 是一种数据序列化系统，它依赖于 Schema。当数据存储时，Schema 也会被存储，这样文件就可以在以后被任何程序读取。这使得 Avro 文件具有良好的可移植性和演化性。

Manifest 文件的组织结构如下：

Snapshot (快照):
- 每个快照文件（JSON 格式）指向一个或多个 Manifest List 文件。
Manifest List (清单列表):
- 存储在 manifest/ 目录下，文件名通常包含 UUID，例如 manifest-list-uuid-N。
- 它是一个 Avro 文件。
- 其内容是 ManifestFileMeta 对象的列表。每个 ManifestFileMeta 描述了一个 Manifest 文件的元数据，包括：
  - Manifest 文件名 (_FILE_NAME)
  - 文件大小 (_FILE_SIZE)
  - 该 Manifest 文件中新增的数据文件数量 (_NUM_ADDED_FILES)
  - 该 Manifest 文件中删除的数据文件数量 (_NUM_DELETED_FILES)
  - 分区统计信息 (_PARTITION_STATS)，用于查询时跳过不相关的 Manifest 文件。
  - 写入此 Manifest 文件时使用的 Schema ID (_SCHEMA_ID)。
- 相关文档：Manifest List 规范
Manifest File (清单文件):
- 存储在 manifest/ 目录下，文件名通常也包含 UUID，例如 manifest-file-uuid。
- 它也是一个 Avro 文件。
- 其内容是 ManifestEntry 对象的列表。每个 ManifestEntry 代表对一个数据文件（Data File）或一个 Changelog 文件的变更记录。
- ManifestEntry 包含以下关键信息：
  - 变更类型 (_KIND): ADD (新增文件) 或 DELETE (删除文件)。
  - 分区值 (_PARTITION): 文件所属的分区。
  - 桶号 (_BUCKET): 文件所属的桶。
  - 数据文件名 (_FILE_NAME)。
  - 数据文件在 LSM 树中的层级 (_LEVEL)。
  - Schema ID (_SCHEMA_ID): 写入此数据文件时使用的 Schema ID。
  - 数据文件的统计信息，如行数、key 的 min/max 值等，用于数据跳过。
  - 关联的索引文件列表 (_EXTRA_FILES)。
- 相关文档：Manifest 规范

由于它们是二进制的 Avro 文件，需要使用 Avro 工具或者通过 Paimon 的代码来读取和解析它们。

通过源码理解 Manifest 文件的创建

主要关注 ManifestFile.java 这个类，它负责 Manifest 文件的创建和读写。

ManifestFile.Factory:

这是创建 ManifestFile 实例的工厂类。
当你需要创建一个新的 Manifest 文件对象（逻辑上的，此时物理文件还未写入或只是准备写入）时，会使用这个工厂。
关键方法是 create()。

ManifestFile.java

// ... existing code ...
public static class Factory {

    private final FileIO fileIO;
    private final SchemaManager schemaManager;
    private final RowType partitionType;
    private final FileFormat fileFormat;
    private final String compression;
    private final FileStorePathFactory pathFactory;
    private final long suggestedFileSize;
    @Nullable private final SegmentsCache<Path> cache;

    public Factory(
            FileIO fileIO,
            SchemaManager schemaManager,
            RowType partitionType,
            FileFormat fileFormat,
            String compression,
            FileStorePathFactory pathFactory,
            long suggestedFileSize,
            @Nullable SegmentsCache<Path> cache) {
        this.fileIO = fileIO;
        this.schemaManager = schemaManager;
        this.partitionType = partitionType;
        this.fileFormat = fileFormat;
        this.compression = compression;
        this.pathFactory = pathFactory;
        this.suggestedFileSize = suggestedFileSize;
        this.cache = cache;
    }

    public ManifestFile create() {
        RowType entryType = VersionedObjectSerializer.versionType(ManifestEntry.SCHEMA);
        return new ManifestFile(
                fileIO,
                schemaManager,
                partitionType,
                new ManifestEntrySerializer(),
                entryType,
                fileFormat.createReaderFactory(entryType),
                fileFormat.createWriterFactory(entryType),
                compression,
                pathFactory.manifestFileFactory(),
                suggestedFileSize,
                cache);
    }
}
// ... existing code ...

在 create() 方法中：

ManifestEntry.SCHEMA 定义了 Manifest 文件中每条记录（ManifestEntry）的 Avro Schema。VersionedObjectSerializer.versionType 可能会对其进行包装以支持版本控制。
fileFormat.createReaderFactory(entryType) 和 fileFormat.createWriterFactory(entryType) 根据配置的 file.format (通常是 Avro) 创建相应的读写器工厂。
pathFactory.manifestFileFactory() 用于生成 Manifest 文件的具体路径和名称。
最终，它调用 ManifestFile 的构造函数来实例化对象。

ManifestFile 构造函数:

接收 Factory 传递过来的参数，初始化 ManifestFile 对象。
这个对象代表一个逻辑上的 Manifest 文件，它知道如何读写物理文件。

ManifestFile.java

// ... existing code ...
public class ManifestFile extends ObjectsFile<ManifestEntry> {

    private final SchemaManager schemaManager;
    private final RowType partitionType;
    private final FormatWriterFactory writerFactory;
    private final long suggestedFileSize;

    private ManifestFile(
            FileIO fileIO,
            SchemaManager schemaManager,
            RowType partitionType,
            ManifestEntrySerializer serializer,
            RowType schema,
            FormatReaderFactory readerFactory,
            FormatWriterFactory writerFactory,
            String compression,
            PathFactory pathFactory,
            long suggestedFileSize,
            @Nullable SegmentsCache<Path> cache) {
        super(
                fileIO,
                serializer,
                schema,
                readerFactory,
                writerFactory,
                compression,
                pathFactory,
                suggestedFileSize,
                cache);
        this.schemaManager = schemaManager;
        this.partitionType = partitionType;
        this.writerFactory = writerFactory;
        this.suggestedFileSize = suggestedFileSize;
    }
// ... existing code ...

ManifestFile.ManifestEntryWriter:

这是一个内部类，继承自 SingleFileWriter。它负责将 ManifestEntry 对象实际写入到物理的 Manifest 文件中。
当 Paimon 的写操作（例如 FileStoreCommitImpl）准备提交一批数据文件的变更时，它会创建一个 ManifestEntryWriter。
通过调用 writer.write(ManifestEntry) 来逐条写入变更记录。
调用 writer.close() 时，会完成文件的写入、关闭流，并返回一个 ManifestFileMeta 对象，该对象描述了刚刚写入的这个 Manifest 文件的元数据。

ManifestFile.java

// ... existing code ...
    @Override
    public ManifestEntrySerializer serializer() {
        return (ManifestEntrySerializer) super.serializer();
    }

    public ManifestEntryWriter createWriter(String fileCompression) {
        return new ManifestEntryWriter(
                writerFactory, pathFactory.newPath(), fileCompression == null ? compression : fileCompression);
    }

    /**
     * Writer for manifest files.
     *
     * <p>IMPORTANT: This writer is not thread-safe.
     */
    public class ManifestEntryWriter extends SingleFileWriter<ManifestEntry, ManifestFileMeta> {

        private final SimpleStatsCollector partitionStatsCollector;
        private final SimpleStatsConverter partitionStatsSerializer;

        private long numAddedFiles = 0;
        private long numDeletedFiles = 0;
        private long schemaId = Long.MIN_VALUE;
        private int minBucket = Integer.MAX_VALUE;
        private int maxBucket = Integer.MIN_VALUE;
        private int minLevel = Integer.MAX_VALUE;
        private int maxLevel = Integer.MIN_VALUE;

        ManifestEntryWriter(FormatWriterFactory factory, Path path, String fileCompression) {
            super(
                    ManifestFile.this.fileIO,
                    factory,
                    path,
                    serializer()::toRow, // 将 ManifestEntry 转换为 Avro 的 GenericRow
                    fileCompression,
                    false);

            this.partitionStatsCollector = new SimpleStatsCollector(partitionType);
            this.partitionStatsSerializer = new SimpleStatsConverter(partitionType);
        }

        @Override
        public void write(ManifestEntry entry) throws IOException {
            super.write(entry); // 调用父类的 write，最终会使用 FormatWriter 写入
            // Update stats
            if (entry.partition().getFieldCount() > 0) {
                partitionStatsCollector.collect(entry.partition());
            }
            if (entry.kind() == FileKind.ADD) {
                numAddedFiles++;
            } else {
                numDeletedFiles++;
            }
            if (schemaId == Long.MIN_VALUE) {
                schemaId = entry.file().schemaId();
            } else if (schemaId != entry.file().schemaId()) {
                // TODO do we need to support different schema ids in one manifest file?
                // if so, we should store a list of schema ids, or store the schema id per
                // entry
                // for now, just check they are the same
                // In the future, if we want to support this, we need to change the
                // ManifestFileMeta.schemaId to a list or remove it.
                // And, AbstractFileStoreScan.readManifestFileStream should also be changed.
                checkState(
                        schemaId == entry.file().schemaId(),
                        "Schema id %s in manifest entry is different from previous schema id %s",
                        entry.file().schemaId(),
                        schemaId);
            }

            minBucket = Math.min(minBucket, entry.bucket());
            maxBucket = Math.max(maxBucket, entry.bucket());
            minLevel = Math.min(minLevel, entry.file().level());
            maxLevel = Math.max(maxLevel, entry.file().level());
        }

        @Override
        public ManifestFileMeta result() throws IOException {
            return new ManifestFileMeta(
                    fileName(),
                    fileSize(),
                    numAddedFiles,
                    numDeletedFiles,
                    partitionStatsSerializer.toBinary(partitionStatsCollector.extract()),
                    schemaId,
                    minBucket,
                    maxBucket,
                    minLevel,
                    maxLevel);
        }
    }
// ... existing code ...

在 ManifestEntryWriter 中：

构造函数接收一个 FormatWriterFactory (例如 AvroWriterFactory) 和目标文件路径。
serializer()::toRow 是一个函数，它将 ManifestEntry Java 对象转换成 Avro 能理解的 InternalRow (或 GenericRow)，然后由底层的 Avro writer 写入文件。
write(ManifestEntry entry) 方法除了调用父类的 write 来实际写入数据外，还会收集统计信息，如新增/删除文件数、分区统计、Schema ID、桶号范围、层级范围等。这些统计信息最终会记录在 ManifestFileMeta 中。
result() 方法在文件写入完成后被调用，用于生成包含上述统计信息的 ManifestFileMeta 对象。

ManifestEntry.java 和 ManifestEntrySerializer.java:
- ManifestEntry.java (通常是一个 POJO 或记录类) 定义了 Manifest 文件中每条记录的逻辑结构。
- ManifestEntrySerializer.java 负责将 ManifestEntry 对象与 Avro 的 InternalRow 进行相互转换。它知道如何从 InternalRow 中读取字段来构造 ManifestEntry，以及如何将 ManifestEntry 的字段写入 InternalRow。

总结

Manifest 文件的创建流程大致是：

当需要记录数据文件变更时（通常在事务提交阶段），Paimon 会通过 ManifestFile.Factory 创建一个 ManifestFile 对象。
然后调用 manifestFile.createWriter(...) 得到一个 ManifestEntryWriter。
将表示数据文件新增或删除的 ManifestEntry 对象逐条写入 ManifestEntryWriter。
ManifestEntryWriter 内部使用 Avro 的 FormatWriter 将 ManifestEntry 序列化为二进制格式并写入磁盘文件。
写入完成后，ManifestEntryWriter 返回一个 ManifestFileMeta，其中包含了这个新生成的 Manifest 文件的元数据。
这些 ManifestFileMeta 对象会被收集起来，写入到 Manifest List 文件中，同样使用 Avro 格式。

通过阅读 ManifestFile.java 中的 Factory 和 ManifestEntryWriter 类，以及相关的 ManifestEntrySerializer.java 和 ManifestEntry.SCHEMA，就能非常清楚地了解 Manifest 文件的创建和内容组织。

3. 数据写入与一致性

总体流程

Flink SQL 是入口: 当使用 Flink SQL INSERT INTO paimon_table ... 时，Flink SQL 的 Planner 会将这个 SQL 语句转换成一个 Flink DataStream 作业。
FlinkTableSink 是桥梁: Paimon 通过实现 DynamicTableSink (即 FlinkTableSink) 来告诉 Flink 如何处理对 Paimon 表的写入。
FlinkSinkBuilder 是构建器: 在 FlinkTableSink.getSinkRuntimeProvider() 中，会使用 FlinkSinkBuilder 来根据表的具体特性（append-only, primary-key, bucket模式等）和配置（log sink, overwrite等）来组装 Sink 逻辑。
FlinkSink 是核心封装: FlinkSinkBuilder 会创建并调用具体 FlinkSink 子类 (如 AppendOnlyFlinkSink, FixedBucketSink) 的 sinkFrom() 方法。
FlinkSink.sinkFrom() 启动流程: 这个方法通过调用 doWrite() 和 doCommit()，将 Paimon 的写入算子 (RowDataStoreWriteOperator) 和提交算子 (CommitterOperator) 通过 Flink DataStream API 的 transform() 方法编织到 Flink 的作业图中。
所以，FlinkSink 及其子类是 Paimon Flink Sink 实现的核心部分，它们负责定义和构建实际执行写入和提交的 Flink 算子。而 Flink SQL 通过 DynamicTableSink 机制间接触发了 FlinkSinkBuilder 和 FlinkSink 的调用，从而将 Paimon 的写入能力集成到 SQL 定义的作业中。用户通常不需要直接与 FlinkSink 交互，除非是在纯 DataStream API 中自定义 Paimon Sink。

数据的写入方式取决于表是否定义了主键。

Append-Only 表 (无主键):
- 写入操作相对简单。新的数据记录会直接追加到新的数据文件中。
- 每次提交会生成新的数据文件，并更新相关的 Manifest 和 Snapshot 元数据。
Primary Key 表 (有主键):
- Paimon 对主键表采用 LSM (Log-Structured Merge-Tree) 树的结构来组织和管理数据文件。这使得 Paimon 能够高效地处理大量的更新 (Update) 和删除 (Delete) 操作。
- 内存中的写缓冲 (Write Buffer):
  - 当数据写入主键表时，记录首先可能被写入内存中的写缓冲区。这个缓冲区通常是排序的。
  - 源码中 MemoryFileStoreWrite (路径: paimon-core/src/main/java/org/apache/paimon/operation/MemoryFileStoreWrite.java) 及其子类负责处理这部分逻辑，它们会使用内存池进行管理。
- 刷盘 (Flush) 与 Sorted Runs:
  - 当写缓冲区达到一定大小或满足其他刷盘条件时，内存中的数据会被刷写到磁盘，形成一个新的数据文件。这个文件内部的记录是根据主键排序的，称为一个 "Sorted Run"。
  - 新生成的 Sorted Run 通常位于 LSM 树的 L0 层。L0 层的文件之间其主键范围可能存在重叠。
- LSM 树的层级 (Levels):
  - LSM 树将 Sorted Run (数据文件) 组织成多个层级 (Level 0, Level 1, Level 2, ...)。
  - L0 层的文件通常是最新写入的，文件较小，且键范围可能重叠。
  - 更高层级 (L1+) 的文件通常由低层级文件合并而来，文件较大，并且在同一层级内，文件之间的主键范围通常不重叠。
  - Levels.java (源码路径: paimon-core/src/main/java/org/apache/paimon/mergetree/Levels.java) 类负责管理这些层级和其中的数据文件 (DataFileMeta)。
- Compaction (合并):
  - 随着数据的不断写入和更新，L0 层的文件数量会增加，或者某些层级的文件变得过于碎片化。这会影响查询性能，因为查询可能需要读取和合并多个文件。
  - Paimon 会自动或通过专门的作业触发 Compaction 操作。Compaction 会选择一个或多个层级中的若干 Sorted Run，将它们合并成新的、更少的、更大的 Sorted Run，并通常将结果写入到更高的层级。
  - 在合并过程中，会处理掉被标记为删除的记录 (DELETE) 和相同主键的旧版本记录 (UPDATE)，只保留最新的有效版本。
- 数据查找 (Lookup):
  - LookupLevels.java (源码路径: org/apache/paimon/mergetree/LookupLevels.java) 文件就是 LSM 树读取和查找逻辑的关键部分。
  - 当需要根据主键查找一条记录时，LookupLevels 会从指定的起始层级开始，逐层遍历 LSM 树中的数据文件。
  - lookup(InternalRow key, int startLevel): 这是查找操作的入口。
  - createLookupFile(DataFileMeta file): 为了加速查找，特别是对于存储在远程文件系统（如 HDFS, S3）上的数据文件，Paimon 可能会为该数据文件在本地创建一个优化的 "lookup file"。
    - 这个 lookup file 本质上是一个本地的键值存储，通常包含布隆过滤器 (BloomFilter) 和对键的索引。
    - 创建过程会读取原始数据文件，使用 keySerializer 序列化键，使用 valueProcessor 处理和序列化值，然后通过 LookupStoreWriter 写入本地文件。bfGenerator 用于生成布隆过滤器。
  - lookupFileCache: 这是一个 Caffeine Cache，用于缓存已经创建的 LookupFile 对象，避免对同一个数据文件重复构建 lookup file，提高后续查找效率。
  - ValueProcessor 接口及其实现 (如 KeyValueProcessor, PositionedKeyValueProcessor): 定义了如何从磁盘读取值字节数组并将其反序列化为目标类型 (T)，以及如何将 KeyValue 持久化到磁盘（在创建 lookup file 时）。例如，PositionedKeyValueProcessor 还会处理行位置信息，这对于 Deletion Vector 等高级功能非常重要。

数据写入

在 Paimon 中，数据写入和提交是一个多阶段的过程，旨在确保数据的一致性和可见性。

1. 谁负责接收请求并实际写入数据？

FlinkSink: 这是 Flink 作业中 Paimon Sink 的入口点和协调者。它负责构建数据写入和提交的 Flink DataStream 转换。
Write Operator (由 createWriteOperatorFactory 创建): FlinkSink 的 doWrite 方法会创建一个特定的写入算子 (Write Operator)。这个算子是 Flink 的一个 OneInputStreamOperator，它在流处理任务中并行运行。此算子接收来自上游的数据流。
StoreSinkWrite (及其实现类，如 StoreSinkWriteImpl): 在 Write Operator 内部，会使用 StoreSinkWrite 的实例。这个组件是实际负责将数据记录写入到物理存储文件（例如 Parquet、ORC 文件）的核心。它处理数据的序列化、文件格式的写入、以及与底层文件系统的交互。

简单来说，Flink 的并行任务实例上运行着 Write Operator，这些 Operator 接收数据，并通过内嵌的 StoreSinkWrite 将数据写入到对应的表文件（data files）中。

代码位置: 在 FlinkSink.java 中，doWrite 方法设置了这个写入流程：

FlinkSink.java

// ...
public DataStream<Committable> doWrite(
        DataStream<T> input, String commitUser, @Nullable Integer parallelism) {
    StreamExecutionEnvironment env = input.getExecutionEnvironment();
    boolean isStreaming = isStreaming(input);

    boolean writeOnly = table.coreOptions().writeOnly();
    // 1. 创建 Write Operator Factory, 其内部会使用 StoreSinkWrite.Provider
    // createWriteOperatorFactory 是一个抽象方法，具体实现由子类提供，例如 AppendOnlyWriterOperatorFactory
    // createWriteProvider 会根据配置选择 StoreSinkWrite 的具体实现，如 StoreSinkWriteImpl
    SingleOutputStreamOperator<Committable> written =
            input.transform(
                    (writeOnly ? WRITER_WRITE_ONLY_NAME : WRITER_WRITE_ONLY_NAME) + " : " + table.name(),
                    new CommittableTypeInfo(),
                    createWriteOperatorFactory( // 此工厂创建的算子负责实际写入
                            createWriteProvider( // 提供 StoreSinkWrite 实例
                                    env.getCheckpointConfig(),
                                    isStreaming,
                                    hasSinkMaterializer(input)),
                            commitUser));
    // ...
    return written;
}

// StoreSinkWrite.Provider 的创建逻辑，最终会返回一个 StoreSinkWrite 的实例
private StoreSinkWrite.Provider createWriteProvider(
        CheckpointConfig checkpointConfig, boolean isStreaming, boolean hasSinkMaterializer) {
    // ...
    // 示例：对于非 write-only 且非 full-compaction 的情况
    if (coreOptions.laziedLookup()) {
        return (table, commitUser, state, ioManager, memoryPool, metricGroup) -> {
            assertNoSinkMaterializer.run();
            return new AsyncLookupSinkWrite( // 这是一个 StoreSinkWrite 实现
                    table,
                    commitUser,
                    state,
                    ioManager,
                    ignorePreviousFiles,
                    waitCompaction,
                    isStreaming,
                    memoryPool,
                    metricGroup);
        };
    }

    return (table, commitUser, state, ioManager, memoryPool, metricGroup) -> {
        assertNoSinkMaterializer.run();
        return new StoreSinkWriteImpl( // 这是另一个 StoreSinkWrite 实现，更常见
                table,
                commitUser,
                state,
                ioManager,
                ignorePreviousFiles,
                waitCompaction,
                isStreaming,
                memoryPool,
                metricGroup);
    };
}

// 子类需要实现这个方法来提供具体的 Writer Operator Factory
protected abstract OneInputStreamOperatorFactory<T, Committable> createWriteOperatorFactory(
        StoreSinkWrite.Provider writeProvider, String commitUser);
// ...

数据写入过程和 Snapshot/Manifest 的创建是紧密相连的，共同构成了 Paimon 的事务提交机制：

数据文件 (Data Files):
- 由上述的 StoreSinkWrite 在 Flink Checkpoint 期间写入。这些是包含实际数据的物理文件（例如 .parquet 文件）。
Committable 对象:
- 当 Write Operator (通过 StoreSinkWrite) 完成一个 Checkpoint 的数据写入后，它会生成一个或多个 Committable 对象。
- Committable (在 Paimon 中通常是 ManifestCommittable) 包含了这次写入操作的元数据信息，例如：
  - 新创建的数据文件的列表和它们的统计信息（行数、大小等）。
  - 对于主键表，可能还包括需要标记为删除的文件（用于更新或删除操作）。
- 这些 Committable 对象会从并行的 Write Operator 发送到下游的单个 Committer Operator。
Committer Operator (提交者):
- FlinkSink 的 doCommit 方法会创建一个 Committer Operator (通常是 CommitterOperatorFactory 创建的 CommitterOperator)。这个算子以单并行度运行。
- 它负责收集来自所有并行 Write Operator 的 Committable 对象。
- 当 Flink 触发 Checkpoint 完成时，Committer Operator 会执行提交操作。这个提交操作由 Committer.Factory (通过 createCommitterFactory() 提供) 创建的 Committer (例如 StoreCommitter) 来执行。
- Committer 的核心职责是：
  1. 创建 Manifest 文件: 根据收集到的 Committable 中的信息，Committer 会生成新的 Manifest 文件。Manifest 文件是一个元数据文件，它列出了属于本次提交的数据文件（新增的、删除的）。
  2. 创建 Manifest List 文件: Manifest List 文件指向一个或多个 Manifest 文件。它将本次提交的所有变更（通过 Manifest 文件描述）组织起来。每个 Snapshot 都会指向一个 Manifest List。
  3. 创建 Snapshot 文件: 这是提交的最后一步，也是原子性的关键。Committer 会创建一个新的 Snapshot 文件（一个 JSON 文件）。这个文件包含了：
    - 当前 Snapshot 的 ID。
    - 本次提交使用的 Schema ID。
    - 指向新创建的 Manifest List 文件的指针。
    - 提交用户、提交类型 (APPEND, COMPACT 等)、提交时间等元数据。
    - 关于本次提交的统计信息，如总记录数、增量记录数等。
原子性: Snapshot 文件的创建（通常是通过文件系统的 rename 操作）是原子性的。一旦 Snapshot 文件成功创建，这次提交写入的数据就对查询可见了。

代码位置: 在 FlinkSink.java 中，doCommit 方法负责提交阶段。

总结:

数据写入阶段 (并行):
- Flink 的并行 Sink Task (Write Operator) 接收数据。
- 每个 Task 内的 StoreSinkWrite 将数据写入到实际的存储介质，生成数据文件 (data files)。
- 每个 Task 在 Checkpoint 时生成 Committable，描述其写入的文件和元数据。
提交阶段 (串行):
- 单个 Committer Operator 收集所有并行 Writer Task 发送过来的 Committable。
- 在 Flink Checkpoint 成功后，Committer Operator 执行提交：
  - 根据 Committable 创建新的 Manifest 文件 (列出数据文件变更)。
  - 创建新的 Manifest List 文件 (指向相关的 Manifest 文件)。
  - 最后，原子地创建新的 Snapshot 文件 (指向 Manifest List，并包含提交元数据)。
- Snapshot 文件的成功创建标志着一批数据的成功提交和对外可见。

这个过程类似于两阶段提交（2PC）的思想，确保了数据写入的原子性和一致性。FlinkSink 扮演了协调者的角色，将 Paimon 核心的写入 (StoreSinkWrite) 和提交 (StoreCommitter，通常由 createCommitterFactory 间接提供)逻辑编排到 Flink 的流式处理框架中。

FlinkSink

FlinkSink 本身并不是一个 Flink 的算子 (Operator)。它更像是一个构建器 (Builder) 或协调器 (Coordinator)，负责将 Paimon 的写入逻辑组装成一个 Flink DataStream 的 Sink 部分。

以下是这个过程：

用户创建 FlinkSink 实例: 用户代码（通常是在 Flink Table API/SQL 层面通过 DynamicTableSink 间接创建，或者直接使用 FlinkSinkBuilder）会创建一个 FlinkSink 的具体子类实例 (例如 AppendOnlyFlinkSink, PrimaryKeyFlinkSink)，并传入 FileStoreTable 对象。
FlinkSink.sinkFrom(DataStream<T> input) 被调用: 这是将 Paimon Sink 连接到 Flink DataStream 的入口点。
FlinkSink.doWrite():
- 此方法接收上游的 DataStream<T>。
- 它调用 createWriteOperatorFactory() (这是一个抽象方法，由 FlinkSink 的子类实现)。这个工厂方法会返回一个 StreamOperatorFactory，例如 RowDataStoreWriteOperator.Factory。
- RowDataStoreWriteOperator.Factory 的构造函数接收 FileStoreTable、可选的 LogSinkFunction、以及一个 StoreSinkWrite.Provider (这个 Provider 是由 FlinkSink.createWriteProvider() 根据表配置创建的)。
- 然后，doWrite() 方法使用 Flink DataStream API 的 transform() 方法，将这个 RowDataStoreWriteOperator.Factory 应用到输入流上。transform() 方法会将这个工厂包装成一个 Flink 的转换操作，当 Flink 作业图构建和执行时，Flink 会使用这个工厂来创建实际的 RowDataStoreWriteOperator 实例，并在 TaskManager 上并行运行它们。
FlinkSink.doCommit():
- doWrite() 返回一个 DataStream<Committable>。
- doCommit() 方法接收这个 Committable 流，并类似地使用 transform() 方法应用一个 CommitterOperatorFactory。这个工厂会创建 CommitterOperator 实例。
- CommitterOperator 负责收集 Committable 并在 Checkpoint 完成后执行 Paimon 的提交逻辑（创建 Manifest 和 Snapshot 文件）。这个算子通常以单并行度运行。

通过这种方式，FlinkSink 将 Paimon 的写入和提交逻辑封装成了标准的 Flink StreamOperatorFactory，使得它们可以无缝地集成到 Flink 的 DataStream 作业图中，并由 Flink 的运行时环境来调度和执行。RowDataStoreWriteOperator 就是这个写入阶段被 Flink 实际执行的核心算子。

总结一下这个链条：

FlinkSink.sinkFrom() -> FlinkSink.doWrite()
doWrite() 调用子类实现的 createWriteOperatorFactory()，并将 createWriteProvider() 的结果（一个 StoreSinkWrite.Provider）传递给它。
子类的 createWriteOperatorFactory() 创建一个 Flink 的 OneInputStreamOperatorFactory (例如 RowDataStoreWriteOperator.Factory)。
这个 Flink Operator Factory 在创建具体的 Flink Operator (例如 RowDataStoreWriteOperator) 时，会调用 StoreSinkWrite.Provider.provide() 方法。
provide() 方法根据 FlinkSink.createWriteProvider() 中的逻辑，实例化一个具体的 StoreSinkWrite 实现 (例如 StoreSinkWriteImpl)。此时，FlinkSink 中持有的 FileStoreTable 实例会被传递给 StoreSinkWrite 的构造函数。
StoreSinkWriteImpl 在其构造函数中，使用传入的 FileStoreTable 调用 table.newWrite(commitUser) 来创建一个 TableWriteImpl 实例。这个 TableWriteImpl 才是真正与 Paimon 核心写逻辑交互的组件。
当 Flink Operator 的 processElement() 方法被调用时，它会调用 StoreSinkWriteImpl.write(rowData)。
StoreSinkWriteImpl.write(rowData) 最终会调用其内部持有的 TableWriteImpl.write(rowData) 方法，将数据写入文件。
在 Checkpoint 时，Flink Operator 调用 StoreSinkWriteImpl.prepareCommit()，后者又调用 TableWriteImpl.prepareCommit() 来生成 ManifestCommittable，其中包含了新写入文件的元数据。

核心流程概览

FlinkSink.doWrite(): 这是发起写入操作的入口。它负责构建 Flink 的 DataStream 转换，将输入数据流转换为 Committable 对象流。
createWriteOperatorFactory() (抽象方法): FlinkSink 是一个抽象类，这个方法需要由其子类实现（例如 FixedBucketSink, CdcFixedBucketSink 等）。此工厂方法创建了一个 Flink 的 OneInputStreamOperatorFactory。这个工厂最终会生产出在 TaskManager 上实际执行数据写入的 Flink 算子 (Operator)。
createWriteProvider(): 在 FlinkSink.doWrite() 内部，会调用 createWriteProvider() 方法。这个方法根据 FileStoreTable 的配置（例如 changelog-producer, write-only 等）来决定使用哪种 StoreSinkWrite.Provider。
StoreSinkWrite.Provider: 这是一个函数式接口，它的 provide() 方法负责实例化一个具体的 StoreSinkWrite 实现。这个 provide() 方法会在 Flink 的写入算子 (Operator) 初始化时被调用。
StoreSinkWrite (例如 StoreSinkWriteImpl): 这是真正执行数据写入逻辑的核心组件。它接收 Flink 算子传递过来的每一条数据，并利用 FileStoreTable 提供的能力将数据写入到存储中。
FileStoreTable 的作用:
- 提供表元数据: StoreSinkWrite 通过 FileStoreTable 对象获取表的 Schema、分区信息、Bucket 定义、核心选项 (CoreOptions) 等。
- 创建 TableWrite 对象: FileStoreTable 有一个关键方法，如 newWrite(commitUser)，它会返回一个 TableWrite 实例 (通常是 TableWriteImpl)。这个 TableWrite 对象封装了向特定 Paimon 表写入数据的底层细节，包括文件格式处理、数据文件组织、索引更新（如果适用）等。
TableWrite (例如 TableWriteImpl): StoreSinkWrite 内部会持有并使用这个 TableWrite 对象来执行具体的写操作，比如调用 tableWrite.write(rowData)。

代码梳理

让我们逐步看代码：

1. FlinkSink.java - 写入的起点

FlinkSink.java

// ... existing code ...
    public DataStream<Committable> doWrite(
            DataStream<T> input, String commitUser, @Nullable Integer parallelism) {
        StreamExecutionEnvironment env = input.getExecutionEnvironment();
        boolean isStreaming = isStreaming(input);

        boolean writeOnly = table.coreOptions().writeOnly();
        // 1. 调用 createWriteOperatorFactory，这是一个抽象方法，由子类实现。
        //    子类的实现会使用 createWriteProvider 返回的 StoreSinkWrite.Provider。
        SingleOutputStreamOperator<Committable> written =
                input.transform(
                        (writeOnly ? WRITER_WRITE_ONLY_NAME : WRITER_NAME) + " : " + table.name(),
                        new CommittableTypeInfo(),
                        createWriteOperatorFactory( // 子类实现此方法
                                createWriteProvider( // 2. createWriteProvider 决定使用哪个 StoreSinkWrite 实现
                                        env.getCheckpointConfig(),
                                        isStreaming,
                                        hasSinkMaterializer(input)),
                                commitUser));
        if (parallelism == null) {
            forwardParallelism(written, input);
        } else {
            written.setParallelism(parallelism);
        }

        Options options = Options.fromMap(table.options());
// ... existing code ...
        if (!table.primaryKeys().isEmpty() && options.get(PRECOMMIT_COMPACT)) {
// ... existing code ...
        }

        return written;
    }
// ... existing code ...
    private StoreSinkWrite.Provider createWriteProvider(
            CheckpointConfig checkpointConfig, boolean isStreaming, boolean hasSinkMaterializer) {
        SerializableRunnable assertNoSinkMaterializer =
                () ->
                        Preconditions.checkArgument(
                                !hasSinkMaterializer,
                                String.format(
                                        "Sink materializer must not be used with Paimon sink. "
                                                + "Please set '%s' to '%s' in Flink's config.",
                                        ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE
                                                .key(),
                                        ExecutionConfigOptions.UpsertMaterialize.NONE.name()));

        Options options = table.coreOptions().toConfiguration();
        ChangelogProducer changelogProducer = table.coreOptions().changelogProducer();
        boolean waitCompaction;
        CoreOptions coreOptions = table.coreOptions();
        if (coreOptions.writeOnly()) {
// ... existing code ...
        }

        // 3. 根据 FileStoreTable 的 coreOptions 决定返回哪种 StoreSinkWrite.Provider
        //    例如，如果配置了 laziedLookup，则返回 AsyncLookupSinkWrite 的 Provider
        if (coreOptions.laziedLookup()) {
            return (table, commitUser, state, ioManager, memoryPool, metricGroup) -> {
                assertNoSinkMaterializer.run();
                // 当 Flink Operator 调用这个 provider 的 provide() 方法时，会创建 AsyncLookupSinkWrite 实例
                return new AsyncLookupSinkWrite(
                        table, // 这个 table 就是 FlinkSink持有的 FileStoreTable
                        commitUser,
                        state,
                        ioManager,
                        ignorePreviousFiles,
                        waitCompaction,
                        isStreaming,
                        memoryPool,
                        metricGroup);
            };
        }

        // 默认情况下，返回 StoreSinkWriteImpl 的 Provider
        return (table, commitUser, state, ioManager, memoryPool, metricGroup) -> {
            assertNoSinkMaterializer.run();
            // 当 Flink Operator 调用这个 provider 的 provide() 方法时，会创建 StoreSinkWriteImpl 实例
            return new StoreSinkWriteImpl(
                    table, // 这个 table 就是 FlinkSink持有的 FileStoreTable
                    commitUser,
                    state,
                    ioManager,
                    ignorePreviousFiles,
                    waitCompaction,
                    isStreaming,
                    memoryPool,
                    metricGroup);
        };
    }

// ... existing code ...
    // 4. 这是一个抽象方法，由 FlinkSink 的子类实现。
    //    例如 FixedBucketSink, CdcFixedBucketSink 等。
    //    子类的实现会创建一个 Flink 的 OneInputStreamOperatorFactory，
    //    这个 Factory 在其内部的 open() 或 createStreamOperator() 方法中，
    //    会调用传入的 writeProvider.provide(...) 来获取 StoreSinkWrite 实例。
    protected abstract OneInputStreamOperatorFactory<T, Committable> createWriteOperatorFactory(
            StoreSinkWrite.Provider writeProvider, String commitUser);

// ... existing code ...

2. FlinkSink 的子类如何实现 createWriteOperatorFactory (以 FixedBucketSink 为例)

FixedBucketSink 是 FlinkWriteSink 的子类，而 FlinkWriteSink 是 FlinkSink 的子类。

FixedBucketSink.java

// ... existing code ...
public class FixedBucketSink extends FlinkWriteSink<InternalRow> {

// ... existing code ...
    public FixedBucketSink(
            FileStoreTable table,
            @Nullable Map<String, String> overwritePartition,
            @Nullable LogSinkFunction logSinkFunction) {
        super(table, overwritePartition);
        this.logSinkFunction = logSinkFunction;
    }

    @Override
    protected OneInputStreamOperatorFactory<InternalRow, Committable> createWriteOperatorFactory(
            StoreSinkWrite.Provider writeProvider, String commitUser) {
        // 5. FixedBucketSink 实现了 createWriteOperatorFactory。
        //    它创建了 RowDataStoreWriteOperator.Factory。
        //    这个 Factory 会在其内部创建 RowDataStoreWriteOperator 实例。
        //    RowDataStoreWriteOperator 在其 open() 方法中会调用 writeProvider.provide()
        //    来获取 StoreSinkWrite 实例，并将其保存在成员变量中，用于后续的 processElement() 调用。
        return new RowDataStoreWriteOperator.Factory(
                table, // 将 FileStoreTable 传递下去
                logSinkFunction,
                writeProvider, // 将 FlinkSink.createWriteProvider() 返回的 Provider 传递下去
                commitUser);
    }
}

`RowDataStoreWriteOperator`

Flink 算子，使用 StoreSinkWrite.Provider

RowDataStoreWriteOperator 是一个 Flink 的流处理算子 (StreamOperator)，其主要作用是：

接收上游数据: 它是一个 OneInputStreamOperator，意味着它从上游 Flink 算子接收数据流。在这个特定的类中，它处理的是 InternalRow 类型的数据，这是 Flink SQL 和 Table API 中常用的内部行数据表示格式。

// ...
public class RowDataStoreWriteOperator extends TableWriteOperator<InternalRow> {
// ...
    @Override
    public void processElement(StreamRecord<InternalRow> element) throws Exception {
        sinkContext.timestamp = element.hasTimestamp() ? element.getTimestamp() : null;

        SinkRecord record;
        try {
            // 调用父类 TableWriteOperator 中持有的 StoreSinkWrite 实例的 write 方法
            record = write.write(element.getValue());
        } catch (Exception e) {
            throw new IOException(e);
        }

        if (record != null
                && logSinkFunction != null
                && (!logIgnoreDelete || record.row().getRowKind().isAdd())) {
            // write to log store, need to preserve original pk (which includes partition fields)
            SinkRecord logRecord = write.toLogRecord(record);
            logSinkFunction.invoke(logRecord, sinkContext);
        }
    }
// ...
}

实际写入数据到 Paimon 表:
- 它继承自 TableWriteOperator。在 TableWriteOperator 的 open() 方法中（或者更早的构造阶段，具体取决于实现），会通过传入的 StoreSinkWrite.Provider 来实例化一个 StoreSinkWrite (例如 StoreSinkWriteImpl)。
- RowDataStoreWriteOperator 的 processElement 方法会调用这个 StoreSinkWrite 实例的 write(InternalRow) 方法，将接收到的 InternalRow 数据实际写入到 Paimon 表的存储文件中（如 Parquet 文件）。

处理 Log Sink (可选):

如果配置了 logSinkFunction (用于将数据变更同步到外部消息队列如 Kafka)，RowDataStoreWriteOperator 会在写入主存储后，将相应的 SinkRecord (可能经过转换以保留原始主键) 发送到 logSinkFunction 进行处理。
它还负责处理 Watermark 并将其传递给 logSinkFunction。

RowDataStoreWriteOperator.java

// ...
public class RowDataStoreWriteOperator extends TableWriteOperator<InternalRow> {
// ...
    @Nullable private final LogSinkFunction logSinkFunction;
    // ...
    @Override
    public void open() throws Exception {
        super.open();

        this.sinkContext = new SimpleContext(getProcessingTimeService());
        if (logSinkFunction != null) {
            openFunction(logSinkFunction);
            logCallback = new LogWriteCallback();
            logSinkFunction.setWriteCallback(logCallback);
            logIgnoreDelete = Options.fromMap(table.options()).get(LOG_IGNORE_DELETE);
        }
    }
// ...
    @Override
    public void processElement(StreamRecord<InternalRow> element) throws Exception {
        // ... (写入主存储的代码) ...

        if (record != null
                && logSinkFunction != null
                && (!logIgnoreDelete || record.row().getRowKind().isAdd())) {
            // write to log store, need to preserve original pk (which includes partition fields)
            SinkRecord logRecord = write.toLogRecord(record);
            logSinkFunction.invoke(logRecord, sinkContext);
        }
    }
// ...
    @Override
    public void processWatermark(Watermark mark) throws Exception {
        super.processWatermark(mark);

        this.currentWatermark = mark.getTimestamp();
        if (logSinkFunction != null) {
            logSinkFunction.writeWatermark(
                    new org.apache.flink.api.common.eventtime.Watermark(mark.getTimestamp()));
        }
    }
// ...
}

生成 Committable:

在 Flink Checkpoint 触发时，RowDataStoreWriteOperator (通过其父类 TableWriteOperator 和 PrepareCommitOperator) 的 prepareCommit 方法会被调用。
这个方法会调用 StoreSinkWrite.prepareCommit() 来获取本次 Checkpoint 写入操作的元数据，封装成 Committable 对象。
如果配置了 logSinkFunction，它还会收集来自 logCallback 的 Log Offset 信息，并将其也封装成 Committable 对象。
这些 Committable 对象会被发送到下游的 Committer 算子。

RowDataStoreWriteOperator.java

// ...
public class RowDataStoreWriteOperator extends TableWriteOperator<InternalRow> {
// ...
    @Override
    protected List<Committable> prepareCommit(boolean waitCompaction, long checkpointId)
            throws IOException {
        // 调用父类的 prepareCommit，它会调用 storeSinkWrite.prepareCommit()
        List<Committable> committables = super.prepareCommit(waitCompaction, checkpointId);

        if (logCallback != null) {
            try {
                Objects.requireNonNull(logSinkFunction).flush();
            } catch (Exception e) {
                throw new IOException(e);
            }
            logCallback
                    .offsets()
                    .forEach(
                            (k, v) ->
                                    committables.add(
                                            new Committable(
                                                    checkpointId,
                                                    Committable.Kind.LOG_OFFSET,
                                                    new LogOffsetCommittable(k, v))));
        }

        return committables;
    }
// ...
}

状态管理: 它还负责管理自身的状态（例如 StoreSinkWriteState，以及 logSinkFunction 的状态），确保在故障恢复时能够正确恢复。

StoreSinkWrite 接口

StoreSinkWrite 接口的核心职责：

写入数据: write(InternalRow rowData) 和 write(InternalRow rowData, int bucket) 是核心的写入方法。
处理 Compaction: compact(...) 用于触发数据文件的合并。
生成 Committable: prepareCommit(...) 在 Flink Checkpoint 时被调用，用于生成包含本次写入元数据的 Committable 对象。
状态管理: snapshotState() 用于在 Checkpoint 时保存自身状态。
生命周期管理: close() 用于释放资源。
Schema 变更支持: replace(FileStoreTable newTable) 用于在 CDC 场景下处理 Schema 变更，通过替换内部的 TableWriteImpl 来适应新的 Schema。

在 FlinkSink.createWriteProvider() 方法中，会根据 FileStoreTable 的配置来决定返回哪种 StoreSinkWrite.Provider。

三个实现类的功能总结

StoreSinkWriteImpl (父类/基础实现)
- 功能：提供了 Paimon Sink 端写入操作的基础框架和通用逻辑。
- 核心职责：
  - 管理底层的 TableWriteImpl 对象，该对象负责实际的数据写入和文件操作。
  - 处理数据写入请求 (write 方法)，将数据路由到正确的 bucket。
  - 处理合并请求 (compact 方法)。
  - 在 Flink checkpoint 时，准备提交物 (prepareCommit 方法)，这些提交物包含了新生成的数据文件和需要更新的元数据信息。
  - 管理 Flink 的状态 (StoreSinkWriteState)，用于在故障恢复时恢复写入进度或相关信息。
  - 处理与内存池、IO 管理器等的交互。
- 特点：它本身不包含特定的高级合并策略，而是提供了一个可扩展的基础，让子类可以实现更复杂的行为。
AsyncLookupSinkWrite (子类/特定场景优化)
- 功能：专门为配置了异步查找（lookup changelog producer 且 LOOKUP_WAIT 为 false）的 Paimon 表设计的 Sink Write。
- 核心职责：
  - 在任务初始化（特别是从 checkpoint 恢复）时，对之前记录为“活跃”（active）的 buckets 执行一次常规合并 (compact(..., false))。
  - 在 Flink checkpoint 时，将当前底层 AbstractFileStoreWrite 认为“活跃”的 buckets 列表保存到 Flink 状态中。
- 特点：
  - 其合并操作是针对性的（只针对活跃 bucket）和即时的（在恢复时立即执行）。
  - 合并类型通常是常规合并，而非全量合并。
  - 目的是确保在异步查找场景下，因异步操作可能导致数据处于中间状态的 bucket 能够快速合并，保证数据一致性或为后续查找准备数据。
GlobalFullCompactionSinkWrite (子类/周期性全局优化)
- 功能：实现周期性的、全局性的全量合并。
- 核心职责：
  - 跟踪记录所有被写入过的 (partition, bucket)，并按 checkpointId 进行组织。
  - 根据 deltaCommits 参数定义的周期，在某个 checkpoint 触发对所有历史写入过的、且尚未被成功全量合并的 buckets 执行一次全量合并 (compact(..., true))。
  - 通过检查 Paimon Snapshot 来确认全量合并操作是否成功提交。
  - 将需要跟踪的写入 bucket 信息持久化到 Flink 状态。
- 特点：
  - 合并操作是全局性的和周期性的。
  - 合并类型是全量合并，更为彻底。
  - 目的是定期对整个表进行深度优化，控制小文件，提升整体性能和数据紧凑性。

`StoreSinkWriteImpl`

StoreSinkWrite 的标准实现，它直接封装了 Paimon 核心的 TableWriteImpl 来执行实际的写操作。

核心功能和实现：

构造与初始化:

在构造时，它接收 FileStoreTable、commitUser、StoreSinkWriteState (用于状态管理)、Flink 的 IOManager、内存池 (MemorySegmentPool 或 MemoryPoolFactory) 以及 MetricGroup 等。
最关键的一步是在构造函数中调用 newTableWrite(table) 方法，该方法会使用传入的 FileStoreTable 实例来创建一个 TableWriteImpl<?> 实例。这个 TableWriteImpl 负责所有与 Paimon 文件存储交互的底层细节。

StoreSinkWriteImpl.java

// ...
public class StoreSinkWriteImpl implements StoreSinkWrite {
// ...
    protected final String commitUser;
    protected final StoreSinkWriteState state; // 用于状态管理
    private final IOManagerImpl paimonIOManager; // Flink IOManager 包装
    // ... 其他成员
    protected TableWriteImpl<?> write; // 核心：持有 TableWriteImpl 实例

    public StoreSinkWriteImpl(
            FileStoreTable table,
            String commitUser,
            StoreSinkWriteState state,
            IOManager ioManager,
            boolean ignorePreviousFiles,
            boolean waitCompaction,
            boolean isStreamingMode,
            @Nullable MemorySegmentPool memoryPool,
            @Nullable MemoryPoolFactory memoryPoolFactory, // CDC 场景可能使用
            @Nullable MetricGroup metricGroup) {
        this.commitUser = commitUser;
        this.state = state;
        this.paimonIOManager = new IOManagerImpl(ioManager);
        this.ignorePreviousFiles = ignorePreviousFiles;
        this.waitCompaction = waitCompaction;
        this.isStreamingMode = isStreamingMode;
        this.memoryPool = memoryPool;
        this.memoryPoolFactory = memoryPoolFactory;
        this.metricGroup = metricGroup;
        // 关键：创建 TableWriteImpl 实例
        this.write = newTableWrite(table);
    }

    private TableWriteImpl<?> newTableWrite(FileStoreTable table) {
        checkArgument(
                !(memoryPool != null && memoryPoolFactory != null),
                "memoryPool and memoryPoolFactory cannot be set at the same time.");

        TableWriteImpl<?> tableWrite =
                table.newWrite( // 使用 FileStoreTable 创建 TableWrite
                                commitUser,
                                (part, bucket) ->
                                        state.stateValueFilter().filter(table.name(), part, bucket))
                        .withIOManager(paimonIOManager)
                        .withIgnorePreviousFiles(ignorePreviousFiles)
                        .withExecutionMode(isStreamingMode)
                        .withBucketMode(table.bucketMode());

        if (metricGroup != null) {
            tableWrite.withMetricRegistry(new FlinkMetricRegistry(metricGroup));
        }

        if (memoryPoolFactory != null) {
            tableWrite.withMemoryPoolFactory(memoryPoolFactory);
        } else {
            tableWrite.withMemoryPool(
                    memoryPool != null
                            ? memoryPool
                            : new HeapMemorySegmentPool( // 默认使用堆内存池
                                    table.coreOptions().writeBufferSize(),
                                    table.coreOptions().pageSize()));
        }

        if (insertOnly != null) {
            tableWrite.withInsertOnly(insertOnly);
        }

        return tableWrite;
    }
// ...
}

数据写入 (write 方法):

write(InternalRow rowData) 和 write(InternalRow rowData, int bucket) 方法直接将调用委托给内部持有的 TableWriteImpl 实例的相应 writeAndReturn 方法。
TableWriteImpl 会处理数据的序列化、分桶、写入到数据文件（例如 Parquet）、以及可能的索引更新等。

StoreSinkWriteImpl.java

// ...
    @Override
    @Nullable
    public SinkRecord write(InternalRow rowData) throws Exception {
        return write.writeAndReturn(rowData); // 委托给 TableWriteImpl
    }

    @Override
    @Nullable
    public SinkRecord write(InternalRow rowData, int bucket) throws Exception {
        return write.writeAndReturn(rowData, bucket); // 委托给 TableWriteImpl
    }
// ...

Compaction (compact 方法):

同样，compact 方法将调用委托给 TableWriteImpl.compact()。

StoreSinkWriteImpl.java

// ...
    @Override
    public void compact(BinaryRow partition, int bucket, boolean fullCompaction) throws Exception {
        write.compact(partition, bucket, fullCompaction); // 委托给 TableWriteImpl
    }
// ...

准备提交 (prepareCommit 方法):

调用 TableWriteImpl.prepareCommit(waitCompaction, checkpointId) 来获取 CommitIncrement。
CommitIncrement 包含了新写入的数据文件 (newFilesIncrement) 和合并产生的文件 (compactIncrement) 的元数据。
然后将这些元数据封装成 Committable 对象列表返回。这些 Committable 会被发送到 Flink 的 Committer 算子。

StoreSinkWriteImpl.java

// ...
    @Override
    public List<Committable> prepareCommit(boolean waitCompaction, long checkpointId)
            throws IOException {
        List<Committable> committables = new ArrayList<>();
        if (write != null) {
            try {
                // 调用 TableWriteImpl 的 prepareCommit
                CommitIncrement increment = write.prepareCommit(waitCompaction, checkpointId);
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Writer {} committed.", ಸ್ಥಾಪನೆ());
                }
                committables.add(
                        new Committable(checkpointId, Committable.Kind.FILE, increment));
            } catch (Exception e) {
                throw new IOException(e);
            }
        }
        return committables;
    }
// ...

状态快照 (snapshotState 方法):

调用 TableWriteImpl.snapshotState() 来获取 TableWriteImpl 内部的状态（例如，当前正在写入的文件、buffer 中的数据等）。
将这些状态通过 StoreSinkWriteState.put() 方法保存起来。StoreSinkWriteState 通常由 StoreSinkWriteStateImpl 实现，它使用 Flink 的 Operator State (如 ListState) 来持久化状态。

StoreSinkWriteImpl.java

// ...
    @Override
    public void snapshotState() throws Exception {
        if (write == null) {
            return;
        }

        List<TableWriteImpl.State<InternalRow>> writeStates = write.checkpoint();
        state.put(
                write.table().name(),
                StoreSinkWriteState.WRITER_STATE_NAME,
                writeStates.stream()
                        .map(
                                s ->
                                        new StoreSinkWriteState.StateValue(
                                                s.partition(),
                                                s.bucket(),
                                                SERIALIZER.serialize(s)))
                        .collect(Collectors.toList()));
    }
// ...

Schema 变更 (replace 方法):

这是为了支持 CDC Sink 在遇到 Schema 变更时能够动态更新写入逻辑。
它会先调用当前 write.checkpoint() 获取旧 TableWriteImpl 的状态。
然后使用新的 FileStoreTable (包含新的 Schema) 创建一个新的 TableWriteImpl 实例。
最后调用新 TableWriteImpl 的 restore() 方法，将旧的状态恢复到新的 TableWriteImpl 中。

StoreSinkWriteImpl.java

// ...
    @Override
    public void replace(FileStoreTable newTable) throws Exception {
        if (write != null) {
            List<TableWriteImpl.State<InternalRow>> states = write.checkpoint();
            write.close();
            write = newTableWrite(newTable);
            write.restore(states);
        } else {
            // Committer may not be initialized
            write = newTableWrite(newTable);
        }
    }
// ...

其他:
- withInsertOnly(): 通知底层的 TableWriteImpl 当前是否为仅插入模式。
- toLogRecord(): 将 SinkRecord 转换为用于 Log Store 的记录。
- close(): 关闭内部的 TableWriteImpl。

总结 StoreSinkWriteImpl: 它是 Paimon Flink Sink 中数据写入和状态管理的核心执行者，通过组合和委托 TableWriteImpl 来完成大部分工作。它为标准的 Paimon 表写入提供了通用的实现。

`GlobalFullCompactionSinkWrite`

GlobalFullCompactionSinkWrite 的核心目标是实现一种周期性的、全局性的全量合并（Full Compaction）机制。

想象一下，在一个持续写入数据的 Paimon 表中，数据会不断地以小文件的形式追加。随着时间的推移，小文件会越来越多，这可能会导致以下问题：

查询性能下降：查询时需要扫描大量的小文件。
元数据管理开销增大：需要跟踪和管理更多的文件。
存储空间利用率不高：小文件可能导致存储空间的碎片化。

Paimon 通过 Compaction（合并）机制来解决这些问题，将小文件合并成更大的文件。而 Full Compaction 是一种更彻底的合并，它会读取一个 bucket 下的所有数据，进行排序和去重（如果表有主键），然后写回新的、更紧凑的文件。

GlobalFullCompactionSinkWrite 的特殊之处在于：

全局性（Globally）：它不是针对单个 bucket 或单个 writer 实例的合并，而是试图在某个时间点（Flink 的 checkpoint）对所有曾经被写入过的 bucket 都触发一次全量合并。
周期性（Periodically）：它不是每次 checkpoint 都触发，而是根据 deltaCommits 参数来决定触发的频率。例如，如果 deltaCommits 设置为 10，那么大约每 10 次 Flink 的 checkpoint commit 之后，它会尝试进行一次全局全量合并。
状态化跟踪（Stateful Tracking）：
- 它会通过 Flink 的状态机制持久化记录在不同 checkpoint 期间，哪些分区（partition）和桶（bucket）被写入过数据。这是通过 writtenBuckets (一个 NavigableMap<Long, Set<Tuple2<BinaryRow, Integer>>>) 实现的，其中 Long 是 checkpointId，Set 包含了该 checkpoint 写入的 (partition, bucket) 对。
- 它还会跟踪哪些 checkpoint 触发了全量合并，并等待确认这些合并操作是否真的成功提交（通过检查 Paimon 的 Snapshot）。这是通过 commitIdentifiersToCheck (一个 TreeSet<Long>) 实现的。

工作流程梳理：

数据写入/常规合并时 (write, compact 方法)：
- 当有数据写入或发生常规合并时，会调用 touchBucket 方法。
- touchBucket 将当前操作涉及的 (partition, bucket) 记录到 currentWrittenBuckets 集合中。
准备提交时 (prepareCommit 方法，在 Flink Checkpoint 触发)：
- 检查已成功的合并：调用 checkSuccessfulFullCompaction()。这个方法会检查 Paimon 的 Snapshot，看之前由这个 SinkWrite 触发的全量合并（其 commit identifier 记录在 commitIdentifiersToCheck 中）是否已经成功生成了 COMPACT 类型的 Snapshot。如果成功，就从 writtenBuckets 和 commitIdentifiersToCheck 中移除相关的记录，表示这些 bucket 已经被成功地全量合并过了，不需要再跟踪它们之前的写入状态了。
- 收集当前写入的 Buckets：将 currentWrittenBuckets（本次 checkpoint 期间写入的 buckets）中的内容添加到 writtenBuckets 中，并以当前 checkpointId 作为键。然后清空 currentWrittenBuckets。
- 判断是否触发全局全量合并：
  - 检查 writtenBuckets 是否为空（即之前是否有过写入）。
  - 检查当前 checkpointId 是否满足 isFullCompactedIdentifier(checkpointId, deltaCommits) 条件。这个条件通常是 checkpointId % deltaCommits == 0 或类似的逻辑，用于控制触发频率。
  - 如果以上两个条件都满足，则将 waitCompaction 标志设置为 true。
- 执行全局全量合并：
  - 如果 waitCompaction 为 true：
    - 调用 submitFullCompaction(checkpointId)。这个方法会遍历 writtenBuckets 中记录的所有历史 checkpoint 期间写入过的所有唯一的 (partition, bucket)，并对每一个都调用 write.compact(partition, bucket, true) 来执行全量合并。
    - 将当前的 checkpointId 添加到 commitIdentifiersToCheck，表示我们触发了一次全量合并，后续需要检查其是否成功。
- 调用父类 prepareCommit：将（可能被修改过的）waitCompaction 标志和 checkpointId 传递给父类，父类会处理实际的提交物（Committable）生成。
状态快照时 (snapshotState 方法，在 Flink Checkpoint 触发)：
- 将 writtenBuckets 中的内容（记录了哪些 checkpoint 写入了哪些 bucket）序列化并保存到 Flink 的状态后端。这样在作业恢复时，可以恢复这些信息，确保不会丢失需要合并的 bucket 记录。

主要实现细节：

继承与构造：

继承 StoreSinkWriteImpl。
构造函数接收 deltaCommits 参数，这个参数决定了大约每隔多少次 Flink 的 checkpoint commit 之后触发一次全局全量合并。
它会从 Flink 的状态（state）中恢复之前记录的 writtenBuckets。writtenBuckets 是一个 NavigableMap<Long, Set<Tuple2<BinaryRow, Integer>>>，key 是 checkpointId，value 是在该 checkpoint 期间被写入过的分区和桶的集合。