This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit f7ab81a155428d9fe7da1e62488fc938c90600e8 Author: Jingsong Lee <[email protected]> AuthorDate: Fri Sep 11 14:17:28 2026 +0800 [docs] Reorganize Learn Paimon guides and add SVG diagrams (#9736) --- docs/docs/learn-paimon/ai-pipelines.mdx | 253 ++++++++ docs/docs/learn-paimon/index.md | 43 ++ docs/docs/learn-paimon/scenario-guide.mdx | 675 +++++----------------- docs/docs/learn-paimon/small-files.mdx | 161 ++++++ docs/docs/learn-paimon/understand-files.mdx | 558 +++++------------- docs/docs/pypaimon/ray-joins.md | 2 +- docs/sidebars.js | 4 +- docs/static/img/learn-paimon-ai-pipeline.svg | 69 +++ docs/static/img/learn-paimon-file-lifecycle.svg | 65 +++ docs/static/img/learn-paimon-small-files.svg | 61 ++ docs/static/img/learn-paimon-streaming-commit.svg | 66 +++ docs/static/img/learn-paimon-table-choice.svg | 55 ++ 12 files changed, 1074 insertions(+), 938 deletions(-) diff --git a/docs/docs/learn-paimon/ai-pipelines.mdx b/docs/docs/learn-paimon/ai-pipelines.mdx new file mode 100644 index 0000000000..26bacf676e --- /dev/null +++ b/docs/docs/learn-paimon/ai-pipelines.mdx @@ -0,0 +1,253 @@ +--- +title: "AI Data Pipelines" +sidebar_position: 2 +--- + +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +# AI Data Pipelines + +Build an AI pipeline by separating payload storage, feature updates, and search indexes. This +guide uses append tables with Data Evolution for managed BLOBs and column updates. Python +readers and some other multimodal capabilities do not require this storage mode. + + + +## Choose the Pieces + +| Requirement | Use | Follow-up | +| --- | --- | --- | +| Store images, audio, documents, or video with metadata | Managed BLOB columns | [BLOB Storage](../multimodal-table/blob) | +| Backfill or replace selected feature columns | Data Evolution | [Partial Updates](../multimodal-table/data-evolution#partial-updates) | +| Store fixed-dimension embeddings separately | VECTOR columns and Vortex files | [Vector Storage](../multimodal-table/vector) | +| Find similar embeddings | Explicitly built vector index | [Vector Index](../multimodal-table/global-index/vector) | +| Filter scalar metadata or search text | BTree or full-text index | [Global Index](../multimodal-table/global-index) | +| Read training batches or run distributed computation | PyPaimon with PyTorch or Ray | [PyPaimon](../pypaimon/) | + +Create Data Evolution tables without a primary key, using the default unaware-bucket layout +(`bucket = -1`), and enable both `row-tracking.enabled` and `data-evolution.enabled`. The `id` +columns below are application join keys; these append tables do not enforce their uniqueness. +See [Data Evolution requirements](../multimodal-table/data-evolution#create-a-table). + +## Store Payloads and Metadata + +The examples assume a configured Paimon catalog. This DDL creates a table used by the Ray +walkthrough below; load uniquely identified payloads before running that walkthrough. + +<Tabs groupId="engine"> +<TabItem value="flink" label="Flink SQL"> + +```sql +CREATE TABLE item_features ( + id BIGINT, + label STRING, + payload BYTES COMMENT '__BLOB_FIELD', + feature BIGINT +) WITH ( + 'row-tracking.enabled' = 'true', + 'data-evolution.enabled' = 'true' +); +``` + +</TabItem> +<TabItem value="spark" label="Spark SQL"> + +```sql +CREATE TABLE item_features ( + id BIGINT, + label STRING, + payload BINARY COMMENT '__BLOB_FIELD', + feature BIGINT +) USING paimon +TBLPROPERTIES ( + 'row-tracking.enabled' = 'true', + 'data-evolution.enabled' = 'true' +); +``` + +</TabItem> +</Tabs> + +BLOB values use dedicated `.blob` files. A query projecting only `id` and `label` can avoid reading +the payload bytes. For ingestion and lazy access to large objects, see +[PyPaimon BLOBs](../pypaimon/blob). JVM primary-key tables have a separate +[BLOB storage design](../primary-key-table/blob-storage) with different constraints. + +## Backfill a Feature Column + +The following **Spark SQL** example is self-contained after catalog setup: + +```sql +CREATE TABLE feature_store ( + user_id BIGINT, + age INT +) USING paimon +TBLPROPERTIES ( + 'row-tracking.enabled' = 'true', + 'data-evolution.enabled' = 'true' +); + +INSERT INTO feature_store VALUES (1, 28), (2, 35); +ALTER TABLE feature_store ADD COLUMNS (score DOUBLE); + +MERGE INTO feature_store AS t +USING (SELECT CAST(1 AS BIGINT) AS user_id, CAST(0.8 AS DOUBLE) AS score) AS s +ON t.user_id = s.user_id +WHEN MATCHED THEN UPDATE SET t.score = s.score; + +SELECT * FROM feature_store ORDER BY user_id; +-- (1, 28, 0.8) +-- (2, 35, NULL) +``` + +The update writes the selected column over each affected normal-file row-ID range. Unmatched +rows in the range keep their existing values. It saves rewriting untouched columns, but does +not imply writing only the one matched row. See +[File Layout and Reads](../multimodal-table/data-evolution-file-layout). + +### Distributed Feature Backfill with Ray + +For expensive feature computation, select the records in Ray, compute a value, and merge only +that column. Install the [Ray integration](../pypaimon/ray-data) first. The join/merge workflow +requires **Ray 2.50 or newer** on the driver and workers. This example assumes: + +- `default.item_features` was created above and populated with unique `id` values. +- The Parquet selection dataset contains only `id`, with one row per selected key and a type + matching the target's `BIGINT`. +- All Ray workers can access the warehouse and input path. Replace the example paths for your environment. +- Resources cover the target payload scan and distributed join as well as inference batches. + This eager example reads payloads before selecting rows through the join. Default BLOB reads + materialize bytes; `batch_size` below limits only the inference batch, not scan or shuffle + memory. Use the [lazy BLOB APIs](../pypaimon/blob) when designing a pipeline for larger objects. + +```python +import ray +from pypaimon.ray import read_paimon, merge_into, WhenMatched, source_col + +catalog_options = {"warehouse": "/path/to/shared/warehouse"} +target = "default.item_features" +num_partitions = 8 # Tune for the cluster and input size. + +records_to_process = ray.data.read_parquet("/path/to/selected-ids/") +target_rows = read_paimon( + target, + catalog_options=catalog_options, + projection=["id", "payload"], +) +selected = records_to_process.join( + target_rows, + join_type="inner", + num_partitions=num_partitions, + on=["id"], +) + + +def compute_feature(batch): + # Byte length is a deterministic stand-in for model inference. + payloads = batch["payload"].to_pylist() + return { + "id": batch["id"].to_pylist(), + "new_feature": [len(value) if value is not None else 0 for value in payloads], + } + + +updates = selected.map_batches( + compute_feature, + batch_format="pyarrow", + batch_size=32, +) +merge_into( + target=target, + source=updates, + catalog_options=catalog_options, + on=["id"], + when_matched=[WhenMatched.update({"feature": source_col("new_feature")})], + num_partitions=num_partitions, +) +``` + +The update source contains keys and new feature values. The merge updates `feature` while +leaving existing BLOB files unchanged. Target normal-file ranges still need alignment, and +selection plus merge can involve distributed joins. See [Ray Joins and Merge](../pypaimon/ray-joins) +for duplicate-match behavior, resource controls, and supported update expressions. + +Coordinate concurrent changes to payloads with feature computation so the result represents the +intended input version. If a workflow persists physical row IDs, also account for +[maintenance that can reassign them](../multimodal-table/data-evolution-maintenance#row-id-lifetime). + +## Store and Search Embeddings + +This **Spark SQL** example uses three-dimensional vectors so the inserts and query are complete. +Configure the [Spark integration](../spark/quick-start) and the +[vector index prerequisites](../multimodal-table/global-index/vector) first. Use your model's +actual dimension in an application. + +```sql +CREATE TABLE doc_embeddings ( + doc_id BIGINT, + title STRING, + embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;3' +) USING paimon +TBLPROPERTIES ( + 'row-tracking.enabled' = 'true', + 'data-evolution.enabled' = 'true', + 'vector.file.format' = 'vortex' +); + +INSERT INTO doc_embeddings VALUES + (1, 'doc_a', array(1.0f, 0.0f, 0.0f)), + (2, 'doc_b', array(0.9f, 0.1f, 0.0f)), + (3, 'doc_c', array(0.0f, 1.0f, 0.0f)); + +CALL sys.create_global_index( + table => 'doc_embeddings', + index_column => 'embedding', + index_type => 'ivf-flat', + options => 'ivf-flat.dimension=3,ivf-flat.distance.metric=cosine,ivf-flat.nlist=1' +); + +SELECT * +FROM vector_search('doc_embeddings', 'embedding', array(1.0f, 0.0f, 0.0f), 2); +``` + +The comment declares a fixed-dimension vector; `vector.file.format = vortex` selects dedicated +vector storage. Building the index is a separate operation after data ingestion. One IVF cluster +is appropriate only for this tiny example. For larger datasets, select index type, training size, +and build/search parameters from the [Vector Index guide](../multimodal-table/global-index/vector). +Vector indexes can also use ordinary `ARRAY<FLOAT>` columns without dedicated vector files. + +New writes do not automatically refresh an existing index. The default `fast` search mode can +omit rows outside index coverage. Decide on a [coverage and freshness policy](../multimodal-table/global-index/manage-indexes#coverage-and-freshness) +before serving queries, and validate recall as well as latency. + +## Read Training and Analysis Data + +Use projection and filtering to avoid loading unused payloads. PyPaimon can feed +[PyTorch](../pypaimon/pytorch), [Ray](../pypaimon/ray-data), and +[Pandas or Arrow](../pypaimon/reading). These reader integrations are not limited to Data +Evolution tables; check the supported table layout and read mode for your pipeline. + +For a reproducible training run, select a retained snapshot or tag and record it with the model +inputs. For continuously updated datasets, make freshness and index coverage explicit. Use +[Understand Files](./understand-files) and [Data Evolution Maintenance](../multimodal-table/data-evolution-maintenance) +to understand the resulting storage and retention costs. diff --git a/docs/docs/learn-paimon/index.md b/docs/docs/learn-paimon/index.md index 68ae077070..88828ac8a7 100644 --- a/docs/docs/learn-paimon/index.md +++ b/docs/docs/learn-paimon/index.md @@ -21,3 +21,46 @@ KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. --> + +# Learn Paimon + +Learn how to choose a table design, follow changes from a writer to storage, and diagnose the +files that remain after an update or compaction. These guides connect Paimon's concepts through +small examples; the linked reference pages cover individual features in detail. + +## Choose a Learning Path + +| Your question | Start here | What you will learn | +| --- | --- | --- | +| Which table fits my workload? | [Scenario Guide](./scenario-guide) | Choose row semantics, distribution, and streaming output separately. | +| How do I store payloads and update AI features? | [AI Data Pipelines](./ai-pipelines) | Combine BLOBs, column updates, vector indexes, and Python processing. | +| What does a write actually change on disk? | [Understand Files](./understand-files) | Trace schema creation, inserts, deletes, compaction, and expiration. | +| Why do I have so many small files? | [Streaming Writes and Small Files](./small-files) | Separate write fragmentation, compaction backlog, and retained history. | + +For a first read, follow the pages in that order. If you already operate a table, start with the +file walkthrough or the small-file diagnosis table. + +## Before You Begin + +Read [Basic Concepts](../concepts/basic-concepts) for snapshots, manifests, partitions, and buckets. +To run examples, configure a Paimon catalog using the [Flink quick start](../flink/quick-start), +[Spark quick start](../spark/quick-start), or [PyPaimon quick start](../pypaimon/quick-start). +Each guide identifies the engine and assumptions for its examples. + +## Keep These Distinctions in Mind + +- **Logical rows and physical records:** a primary-key query can return one row while several + versions of that key remain in data files. +- **Commit and compaction:** a commit publishes a snapshot; compaction reorganizes files and + publishes its result through a commit. +- **Compaction and expiration:** retiring a file from the current snapshot does not immediately + remove it from storage. Retained history can still need it. +- **Storage and indexes:** storing a vector or text value does not automatically build a search + index. Index coverage has its own lifecycle. + +## Continue with the References + +Use [Primary-Key Tables](../primary-key-table/) for merge and changelog behavior, +[Append Tables](../append-table/) for batch and ordered append layouts, and +[Multimodal Tables](../multimodal-table/) for column storage and search. For ongoing operations, +see [Maintenance](../maintenance/) and [System Tables](../concepts/system-tables). diff --git a/docs/docs/learn-paimon/scenario-guide.mdx b/docs/docs/learn-paimon/scenario-guide.mdx index bc0d0f5e29..337b37ffea 100644 --- a/docs/docs/learn-paimon/scenario-guide.mdx +++ b/docs/docs/learn-paimon/scenario-guide.mdx @@ -1,11 +1,8 @@ --- title: "Scenario Guide" -sidebar_position: 2 +sidebar_position: 1 --- -import Tabs from '@theme/Tabs'; -import TabItem from '@theme/TabItem'; - <!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file @@ -27,149 +24,132 @@ under the License. # Scenario Guide -This guide helps you choose the right Paimon table type and configuration for your specific use case. Paimon provides -**Primary Key Table**, **Append Table**, and **Multimodal Data Lake** capabilities — each with different modes and -configurations that are suited for different scenarios. +Choose a table by what incoming records mean. Then choose distribution, read behavior, and +streaming output. A need for SQL `UPDATE` or `DELETE` alone does not require a primary-key table: +append tables also support engine-specific row-level operations. + +<a id="summary-table-type-decision-tree"></a> + + ## Quick Decision -| Scenario | Table Type | Key Configuration | -|---|---|---| -| CDC real-time sync from database | Primary Key Table | `deletion-vectors.enabled = true` | -| Streaming aggregation / metrics | Primary Key Table | `merge-engine = aggregation` | -| Multi-stream partial column updates | Primary Key Table | `merge-engine = partial-update` | -| Log deduplication (keep first) | Primary Key Table | `merge-engine = first-row` | -| Batch ETL / data warehouse layers | Append Table | Default (unaware-bucket) | -| High-frequency point queries on key | Append Table | `bucket = N, bucket-key = col` | -| Queue-like ordered streaming | Append Table | `bucket = N, bucket-key = col` | -| Large-scale OLAP with ad-hoc queries | Append Table | Incremental Clustering | -| Store images / videos / documents | Append Table (Blob) | `__BLOB_FIELD` comment, Data Evolution enabled | -| AI vector search / RAG | Append Table (Vector) | `VECTOR` type, Vector Global Index | -| AI feature engineering & column evolution | Append Table | `data-evolution.enabled = true` | -| Python AI pipeline (Ray / PyTorch) | Append Table | PyPaimon SDK | +| Workload | Starting point | Check before adopting it | +| --- | --- | --- | +| CDC or upsert replication | Primary-key table, default `deduplicate` merge engine | Source ordering, partition changes, and downstream changelog requirements | +| Independent updates to columns of one entity | Primary-key table, `partial-update` | Per-source sequence groups and null/delete semantics | +| Accumulating metric contributions | Primary-key table, `aggregation` | Input replay and each function's retraction behavior | +| Keep the first record received for a key | Primary-key table, `first-row` | Arrival order differs from earliest event time | +| Batch ETL, logs, or partition overwrite | Unaware-bucket append table | Partition granularity and file sizes | +| Equality filters or compatible bucketed joins | Fixed-bucket append table | Complete bucket-key predicates, skew, and the query plan | +| Preserve append order within a partition and bucket | Ordered bucketed append table | No global or event-time ordering guarantee | +| BLOBs, feature backfills, or vector search | [AI Data Pipelines](./ai-pipelines) | Storage requirements and index freshness are separate decisions | ---- +The SQL examples below use **Flink SQL** in an existing Paimon catalog. Bucket counts are small +illustrative values, not sizing recommendations. For Spark syntax, use the linked feature guides. ## Primary Key Table -Use a Primary Key Table when your data has a natural unique key and you need **real-time updates** (insert, update, delete). -See [Primary Key Table Overview](../primary-key-table/). - -### Scenario 1: CDC Real-Time Sync +Use a [primary-key table](../primary-key-table/) when incoming records should merge into one +logical row per key. Pick the merge engine before tuning the storage mode. -**When:** You want to synchronize a MySQL / PostgreSQL / MongoDB table to the data lake in real-time with upsert -semantics. This is the most common use case for Primary Key Tables. +### CDC Real-Time Sync {#scenario-1-cdc-real-time-sync} -**Recommended Configuration:** +For a source that supplies a monotonically increasing version per key: ```sql CREATE TABLE orders ( order_id BIGINT, - user_id BIGINT, - amount DECIMAL(10,2), + amount DECIMAL(12, 2), status STRING, - update_time TIMESTAMP, - dt STRING, - PRIMARY KEY (order_id, dt) NOT ENFORCED -) PARTITIONED BY (dt) WITH ( - 'deletion-vectors.enabled' = 'true', - 'changelog-producer' = 'lookup', - 'sequence.field' = 'update_time' + source_version BIGINT, + PRIMARY KEY (order_id) NOT ENFORCED +) WITH ( + 'sequence.field' = 'source_version' ); ``` -**Why this configuration:** +The default `deduplicate` engine keeps the row with the largest sequence value. Use a source +version that is non-null and expresses the required precedence; timestamps can tie. A sequence field alone does +not resolve equal versions deterministically. See [Sequence Field](../primary-key-table/sequence-rowkind#sequence-field). -- **`deletion-vectors.enabled = true`** (MOW mode): Enables [Merge On Write](../primary-key-table/table-mode#merge-on-write) with Deletion Vectors. - This mode gives you the best balance of write and read performance. Compared to the default MOR mode, MOW - avoids merging at read time, which greatly improves OLAP query performance. -- **`changelog-producer = lookup`**: Generates a complete [changelog](../primary-key-table/changelog-producer#lookup) - for downstream streaming consumers. If your CDC source is directly connected to a database (e.g., MySQL CDC, Postgres CDC), - you can use `changelog-producer = input` instead, since the database CDC stream already provides a complete changelog. - However, if your CDC source comes from Kafka (or other message queues), `input` may not be reliable — use `lookup` to - ensure changelog correctness. If no downstream streaming read is needed, you can omit this to save compaction resources. -- **`sequence.field = update_time`**: Guarantees correct update ordering even when data arrives out of order. -- **Bucketing**: Use the default Dynamic Bucket (`bucket = -1`). The system automatically adjusts bucket count based - on data volume. If you are sensitive to data visibility latency, set a fixed bucket number (e.g. `'bucket' = '5'`) - — roughly 1 bucket per 1GB of data in a partition. +Choose the [changelog producer](../primary-key-table/changelog-producer) according to the +records reaching Paimon and the consumer's needs: -**CDC Ingestion Tip:** Use [Paimon CDC Ingestion](../cdc-ingestion/) for whole-database sync with -automatic table creation and schema evolution support. +| Consumer/input contract | Producer | +| --- | --- | +| Batch reads only, or consumers can handle upserts | Leave the default `none`. | +| Input already contains the complete required before/after changes | Use `input`. | +| Input lacks before images, but consumers need complete changes | Consider `lookup`; budget for lookup compaction. | -### Scenario 2: Multi-Stream Partial Column Updates +The transport does not determine changelog completeness: a Kafka stream can carry complete CDC, +and a database connector can omit information required by a particular consumer. Partial-update +and aggregation inputs also differ from their final merged rows. -**When:** Multiple data sources each contribute different columns to the same record, and you want to progressively -merge them into a complete wide table (e.g., orders from one stream + logistics info from another). +If analytical reads dominate, evaluate [MOW with deletion vectors](../primary-key-table/table-mode#merge-on-write) +at table creation. Check its compaction and visibility requirements instead of enabling it as a +universal CDC default. For whole-database ingestion and schema evolution, see [CDC Ingestion](../cdc-ingestion/). -**Recommended Configuration:** +If you partition a primary-key table, the partition columns normally belong to its primary key. +An entity moving between partitions needs an explicit design; see +[Cross-Partition Upsert](../primary-key-table/data-distribution#cross-partitions-upsert). + +### Partial Column Updates {#scenario-2-multi-stream-partial-column-updates} + +For two sources that update different parts of an order: ```sql CREATE TABLE order_wide ( order_id BIGINT PRIMARY KEY NOT ENFORCED, - -- from order stream - user_name STRING, - amount DECIMAL(10,2), - order_time TIMESTAMP, - -- from logistics stream - tracking_no STRING, + amount DECIMAL(12, 2), + order_version BIGINT, delivery_status STRING, - delivery_time TIMESTAMP + delivery_version BIGINT ) WITH ( 'merge-engine' = 'partial-update', - 'fields.order_time.sequence-group' = 'user_name,amount', - 'fields.delivery_time.sequence-group' = 'tracking_no,delivery_status', - 'deletion-vectors.enabled' = 'true', + 'fields.order_version.sequence-group' = 'amount', + 'fields.delivery_version.sequence-group' = 'delivery_status', 'changelog-producer' = 'lookup' ); ``` -**Why:** The [partial-update](../primary-key-table/merge-engine/partial-update) merge engine allows each -stream to update only its own columns without overwriting the others. `sequence-group` ensures ordering within each -stream independently. +Each source supplies its own version and leaves the other group's version null. A null sequence +skips that group; an accepted group update can set its value fields to null. Fields outside +sequence groups use the default non-null update behavior. `lookup` serves consumers that need +complete merged rows. See [Partial Update](../primary-key-table/merge-engine/partial-update). -### Scenario 3: Streaming Aggregation / Metrics +Combine the input streams into one writing job when using the default dynamic buckets. Multiple +sources do not imply that multiple independent writers can safely assign dynamic buckets. -**When:** You need to pre-aggregate metrics in real-time (e.g., page views, total sales, UV count). Each incoming record -should be aggregated with the existing value, not replace it. +### Streaming Metrics {#scenario-3-streaming-aggregation--metrics} -**Recommended Configuration:** +Use aggregation when each input is a contribution, such as an additional quantity sold: ```sql CREATE TABLE product_metrics ( - product_id BIGINT, - dt STRING, - total_sales BIGINT, - max_price DOUBLE, - uv VARBINARY, - PRIMARY KEY (product_id, dt) NOT ENFORCED -) PARTITIONED BY (dt) WITH ( + product_id BIGINT PRIMARY KEY NOT ENFORCED, + quantity BIGINT, + max_price DOUBLE +) WITH ( 'merge-engine' = 'aggregation', - 'fields.total_sales.aggregate-function' = 'sum', + 'fields.quantity.aggregate-function' = 'sum', 'fields.max_price.aggregate-function' = 'max', - 'fields.uv.aggregate-function' = 'hll_sketch', - 'deletion-vectors.enabled' = 'true', 'changelog-producer' = 'lookup' ); ``` -**Why:** The [aggregation](../primary-key-table/merge-engine/aggregation) merge engine supports 20+ -aggregate functions (`sum`, `max`, `min`, `count`, `hll_sketch`, `theta_sketch`, `collect`, `merge_map`, etc.), ideal -for real-time metric accumulation. - -### Scenario 4: Log Deduplication (Keep First Record) - -**When:** You receive a high-volume log stream with possible duplicates and only want to keep the first occurrence of -each key (e.g., first login event per user per day). +Replaying a contribution can add it again. Do not send full replacement totals to a `sum` field +unless that is the intended calculation. Functions differ in support for retracts, deletes, and +input encoding; sketches require prepared sketch values. See +[Aggregation](../primary-key-table/merge-engine/aggregation). -**Recommended Configuration:** +### Keep the First Record {#scenario-4-log-deduplication-keep-first-record} ```sql CREATE TABLE first_login ( user_id BIGINT, dt STRING, - login_time TIMESTAMP, - device STRING, - ip STRING, + login_time TIMESTAMP(3), PRIMARY KEY (user_id, dt) NOT ENFORCED ) PARTITIONED BY (dt) WITH ( 'merge-engine' = 'first-row', @@ -177,479 +157,118 @@ CREATE TABLE first_login ( ); ``` -**Why:** The [first-row](../primary-key-table/merge-engine/first-row) merge engine keeps only the earliest -record for each primary key and produces insert-only changelog, making it perfect for streaming log deduplication. +This keeps the first row received for a key, not necessarily the row with the earliest +`login_time`. Lookup compaction produces an insert-only changelog of newly retained keys. +User-defined sequence fields and ordinary row deletes are not supported. Do not add deletion +vectors to this configuration; see [First Row](../primary-key-table/merge-engine/first-row). -### Primary Key Table: Bucket Mode Comparison +### Choose Buckets and Read Behavior {#primary-key-table-bucket-mode-comparison} -| Mode | Config | Best For | Trade-off | -|---|---|---|---| -| Dynamic Bucket (default) | `bucket = -1` | Most scenarios, auto-scaling | Requires single write job | -| Fixed Bucket | `bucket = N` | Stable workloads, bucketed join | Manual rescaling needed | -| Postpone Bucket | `bucket = -2` | Adaptive Partition Level Bucket | New data not visible until compaction | +| Bucket mode | Configuration | Main consideration | +| --- | --- | --- | +| Dynamic (default) | `bucket = -1` | Assignment adapts to data; use one writing job. | +| Fixed | Positive `bucket` | Plan for skew, parallelism, file sizes, and later rescaling. | +| Postpone | `bucket = -2` | New data needs compaction before it becomes queryable. | -**General guideline:** ~1 bucket per 1GB of data in a partition, with each bucket containing 200MB–1GB of data. +Use [Data Distribution](../primary-key-table/data-distribution) to size and validate the layout. +A single bytes-per-bucket rule cannot capture update rate, partition count, or available memory. -### Primary Key Table: Table Mode Comparison +#### Table Mode {#primary-key-table-table-mode-comparison} -| Mode | Config | Write Perf | Read Perf | Best For | -|---|---|---|---|---| -| MOR (default) | — | Very good | Not so good | Write-heavy, less query | -| COW | `full-compaction.delta-commits = 1` | Very bad | Very good | Read-heavy, batch jobs | -| MOW | `deletion-vectors.enabled = true` | Good | Good | Balanced (recommended for most) | +| Mode | Configuration | Work to budget for | +| --- | --- | --- | +| MOR (default) | No extra setting | Readers merge overlapping versions; compaction reduces that work. | +| COW | `full-compaction.delta-commits = 1` | Frequent full compaction increases write amplification. | +| MOW | `deletion-vectors.enabled = true` | Writers resolve versions through lookup compaction; readers apply deletion vectors. | ---- +These are [read/write trade-offs](../primary-key-table/table-mode), not interchangeable presets. +In particular, `lookup` changelog generation cannot be combined with `full-compaction.delta-commits`, +and deletion-vector tables do not support the `full-compaction` changelog producer. ## Append Table -Use an Append Table when your data **has no natural primary key**, or you are working with **batch ETL** pipelines -where data is only inserted and does not need upsert semantics. -See [Append Table Overview](../append-table/). - -Compared to Primary Key Tables, Append Tables have much better batch read/write performance, simpler design, and lower -resource consumption. **We recommend using Append Tables for most batch processing scenarios.** - -### Scenario 5: Batch ETL / Data Warehouse Layers (Unaware-Bucket) - -**When:** Standard data warehouse layering (ODS → DWD → DWS → ADS), bulk INSERT OVERWRITE, or Spark / Hive style -batch processing. +Use an [append table](../append-table/) when each input is stored independently, or when your +engine manages batch replacements and row-level changes. A business identifier in the schema +does not automatically need to be a declared primary key. -**Recommended Configuration:** +### Batch ETL and Analytical Queries {#scenario-5-batch-etl--data-warehouse-layers-unaware-bucket} ```sql -CREATE TABLE dwd_events ( +CREATE TABLE events ( event_id BIGINT, user_id BIGINT, event_type STRING, - event_time TIMESTAMP, dt STRING ) PARTITIONED BY (dt); ``` -No bucket configuration needed. This is an **unaware-bucket** append table — the simplest and most commonly used form. -Paimon automatically handles small file merging and supports: - -- **Time travel** and version rollback. -- **Schema evolution** (add/drop/rename columns). -- **Data skipping** via min-max stats in manifest files. -- **File Index** (BloomFilter, Bitmap, Range Bitmap) for further query acceleration. -- **Row-level operations** (DELETE / UPDATE / MERGE INTO in Spark SQL). -- **Incremental Clustering** for advanced data layout optimization. - -**Query Optimization Tip:** If your queries frequently filter on specific columns, consider using -[Incremental Clustering](../append-table/incremental-clustering) to sort data by those columns: - -```sql -ALTER TABLE dwd_events SET ( - 'clustering.incremental' = 'true', - 'clustering.columns' = 'user_id' -); -``` - -Or define a File Index for point lookups: - -```sql -ALTER TABLE dwd_events SET ( - 'file-index.bloom-filter.columns' = 'user_id' -); -``` - -### Scenario 6: High-Frequency Point Queries (Bucketed Append) +With no primary key and no bucket setting, this uses the unaware-bucket layout. Start here for +batch inserts and partition overwrite. Add partition filters and inspect file statistics before +adding layout options. For selective queries, evaluate [File Indexes](../append-table/query-performance) +or [Incremental Clustering](../append-table/incremental-clustering). Supported SQL mutations and +their costs are described in [Row-Level Operations](../append-table/row-level-operations). -**When:** Your append table is frequently queried with equality or IN filters on a specific column -(e.g., `WHERE product_id = xxx`). This is the **most impactful** advantage of a bucketed append table. - -**Recommended Configuration:** +### Equality Lookups and Bucketed Joins {#scenario-6-high-frequency-point-queries-bucketed-append} ```sql CREATE TABLE product_logs ( product_id BIGINT, - log_time TIMESTAMP, - message STRING, - dt STRING -) PARTITIONED BY (dt) WITH ( - 'bucket' = '16', + log_time TIMESTAMP(3), + message STRING +) WITH ( + 'bucket' = '8', 'bucket-key' = 'product_id' ); -``` - -**Why this is powerful:** The `bucket-key` enables **data skipping** — when a query contains `=` or `IN` conditions on -the bucket-key, Paimon pushes these predicates down and prunes all irrelevant bucket files entirely. With 16 buckets, -a point query on `product_id` only reads ~1/16 of the data. -```sql --- Only reads the bucket containing product_id=12345, skips all other 15 buckets SELECT * FROM product_logs WHERE product_id = 12345; - --- Only reads buckets for these 3 values -SELECT * FROM product_logs WHERE product_id IN (1, 2, 3); ``` -See [Bucketed Append — Data Skipping](../append-table/bucketed#data-skipping). +An equality or `IN` predicate on the complete bucket key can prune unrelated buckets. This does +not guarantee reading exactly one eighth of the bytes: bucket sizes can be skewed, and matching +buckets still contain other keys. A composite bucket key requires constraints on every key column. -If queries also frequently filter on columns other than the `bucket-key` and ordered streaming is not required, you can -enable [Incremental Clustering](../append-table/incremental-clustering) on the bucketed append table by setting -`bucket-append-ordered = false`. +Spark can exploit compatible bucket distributions in joins with V2 bucketing enabled. Confirm +shuffle removal with `EXPLAIN`; see [Bucketed Join](../append-table/bucketed#bucketed-join). -**Bucketed Join Bonus:** If two bucketed tables share the same `bucket-key` and bucket count, Spark can join them -**without shuffle**, significantly accelerating batch join queries: +### Ordered Streaming {#scenario-7-queue-like-ordered-streaming-bucketed-append} -```sql -SET spark.sql.sources.v2.bucketing.enabled = true; - --- Both tables have bucket=16, bucket-key=product_id -SELECT * FROM product_logs JOIN product_dim -ON product_logs.product_id = product_dim.product_id; -``` - -See [Bucketed Join](../append-table/bucketed#bucketed-join). +The `product_logs` table above also preserves append order within each partition and bucket +because `bucket-append-ordered` defaults to `true`. It does not sort by `log_time` or define a +global order across buckets. See [Bucketed Streaming](../append-table/bucketed#bucketed-streaming). -### Scenario 7: Queue-Like Ordered Streaming (Bucketed Append) +Bucketed incremental clustering requires `bucket-append-ordered = false`, giving up that order +guarantee. Decide whether ordering or clustering is the requirement before combining the options. -**When:** You want to use Paimon as a message queue replacement with strict ordering guarantees per key (similar to -Kafka partitioning), with the benefits of filter push-down and lower cost. +### Compare Append Layouts {#append-table-bucket-mode-comparison} -**Recommended Configuration:** - -```sql -CREATE TABLE event_stream ( - user_id BIGINT, - event_type STRING, - event_time TIMESTAMP(3), - payload STRING, - WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND -) WITH ( - 'bucket' = '8', - 'bucket-key' = 'user_id' -); -``` - -**Why:** Within the same bucket, records are strictly ordered by write time. Streaming reads deliver records in exact -write order per bucket. This gives you Kafka-like partitioned ordering at data lake cost. - -See [Bucketed Streaming](../append-table/bucketed#bucketed-streaming). - -### Append Table: Bucket Mode Comparison - -| Mode | Config | Data Skipping | Bucketed Join | Ordered Streaming | Incremental Clustering | -|---|---|---|---|---|---| -| Unaware-Bucket (default) | No bucket config | Via min-max / file index | No | No | Yes | -| Bucketed | `bucket = N, bucket-key = col` | **Bucket-key filter pushdown** | Yes | Yes, unless `bucket-append-ordered = false` | Yes, requires `bucket-append-ordered = false` | - ---- +| Requirement | Unaware-bucket | Fixed-bucket | +| --- | --- | --- | +| Bucket-key pruning and compatible bucketed joins | No | Yes, with compatible keys and query plans | +| Append order within a partition and bucket | Not guaranteed | Yes, when `bucket-append-ordered = true` | +| Incremental clustering | Supported | Requires `bucket-append-ordered = false` | +| Data Evolution | Supported with its required options | Not supported | ## Multimodal Data Lake -Paimon is a multimodal lakehouse for AI. You can keep multimodal data, metadata, and embeddings in the same table and -query them via vector search, full-text search, or SQL. All multimodal features are built on top of Append Tables with -[Data Evolution](../multimodal-table/data-evolution) mode enabled. - -### Scenario 8: Storing Multimodal Data (Blob Table) - -**When:** You need to store images, videos, audio files, documents, or model weights alongside structured metadata in -the data lake, and want efficient column projection without loading large binary data. - -**Recommended Configuration:** - -<Tabs groupId="blob-scenario"> - -<TabItem value="flink-sql" label="Flink SQL"> - -```sql -CREATE TABLE image_table ( - id INT, - name STRING, - label STRING, - image BYTES COMMENT '__BLOB_FIELD' -) WITH ( - 'row-tracking.enabled' = 'true', - 'data-evolution.enabled' = 'true' -); -``` - -</TabItem> - -<TabItem value="spark-sql" label="Spark SQL"> - -```sql -CREATE TABLE image_table ( - id INT, - name STRING, - label STRING, - image BINARY COMMENT '__BLOB_FIELD' -) TBLPROPERTIES ( - 'row-tracking.enabled' = 'true', - 'data-evolution.enabled' = 'true' -); -``` - -</TabItem> - -</Tabs> - -**Why:** The [Blob Storage](../multimodal-table/blob) separates large binary data into dedicated `.blob` files -while metadata stays in standard columnar files (Parquet/ORC). This means: - -- `SELECT id, name, label FROM image_table` does **not** load any blob data — very fast. -- Blob data supports streaming reads for large objects (videos, model weights) without loading entire files into memory. -- Supports multiple input methods: local files, HTTP URLs, InputStreams, and byte arrays. - -### Scenario 9: Vector Search / RAG Applications - -**When:** You are building a recommendation system, image retrieval, or RAG (Retrieval Augmented Generation) -application that needs approximate nearest neighbor (ANN) search on embeddings. - -**Recommended Configuration:** - -<Tabs groupId="vector-scenario"> - -<TabItem value="spark-sql" label="Spark SQL"> - -```sql -CREATE TABLE doc_embeddings ( - doc_id INT, - title STRING, - content STRING, - embedding ARRAY<FLOAT> COMMENT '__VECTOR_FIELD;768' -) TBLPROPERTIES ( - 'row-tracking.enabled' = 'true', - 'data-evolution.enabled' = 'true', - 'global-index.enabled' = 'true', - 'vector.file.format' = 'lance' -); -``` - -</TabItem> - -<TabItem value="java-api" label="Java API"> - -```java -Schema schema = Schema.newBuilder() - .column("doc_id", DataTypes.INT()) - .column("title", DataTypes.STRING()) - .column("content", DataTypes.STRING()) - .column("embedding", DataTypes.VECTOR(768, DataTypes.FLOAT())) - .option("row-tracking.enabled", "true") - .option("data-evolution.enabled", "true") - .option("global-index.enabled", "true") - .option("vector.file.format", "lance") - .build(); -``` - -</TabItem> - -</Tabs> - -**Build the vector index and search:** - -```sql --- Build IVF-PQ vector index -CALL sys.create_global_index( - table => 'db.doc_embeddings', - index_column => 'embedding', - index_type => 'ivf-pq', - options => 'vector.distance.metric=cosine,vector.nlist=256,vector.pq.m=16' -); - --- Search for top-5 nearest neighbors -SELECT * FROM vector_search('doc_embeddings', 'embedding', array(0.1f, 0.2f, ...), 5); -``` +For payload storage, embedding search, feature backfills, and Python training readers, continue +with [AI Data Pipelines](./ai-pipelines). That guide separates storage, column updates, and indexes, +and provides a complete small vector-search example. -**Why:** The [Global Index](../multimodal-table/global-index) with vector indexes provides high-performance ANN search. -Vector data is stored in dedicated `.vector.lance` files optimized for dense vectors, while scalar columns stay in -Parquet. You can also build a **BTree Index** on scalar columns for efficient filtering: +| Pipeline task | Walkthrough | +| --- | --- | +| <a id="scenario-8-storing-multimodal-data-blob-table"></a>Store payloads | [BLOBs and metadata](./ai-pipelines#store-payloads-and-metadata) | +| <a id="scenario-9-vector-search--rag-applications"></a>Search embeddings | [Vector storage and search](./ai-pipelines#store-and-search-embeddings) | +| <a id="scenario-10-ai-feature-engineering-with-data-evolution"></a>Update features | [Column backfill](./ai-pipelines#backfill-a-feature-column) | +| <a id="scenario-11-python-ai-pipeline-pypaimon"></a>Train or analyze in Python | [Training readers](./ai-pipelines#read-training-and-analysis-data) | -```sql --- Build BTree index for scalar filtering -CALL sys.create_global_index( - table => 'db.doc_embeddings', - index_column => 'title', - index_type => 'btree' -); - --- Scalar lookup is accelerated by BTree index -SELECT * FROM doc_embeddings WHERE title IN ('doc_a', 'doc_b'); -``` - -### Scenario 10: AI Feature Engineering with Data Evolution - -**When:** You have a feature store or data pipeline where new feature columns are added frequently, and you want to -backfill or update specific columns without rewriting entire data files. - -**Recommended Configuration:** - -```sql -CREATE TABLE feature_store ( - user_id INT, - age INT, - gender STRING, - purchase_count INT -) TBLPROPERTIES ( - 'row-tracking.enabled' = 'true', - 'data-evolution.enabled' = 'true' -); -``` - -**Update only specific columns via MERGE INTO:** - -```sql --- Later, add a new feature column -ALTER TABLE feature_store ADD COLUMNS (embedding ARRAY<FLOAT>); - --- Backfill only the new column — no full file rewrite! -MERGE INTO feature_store AS t -USING embedding_source AS s -ON t.user_id = s.user_id -WHEN MATCHED THEN UPDATE SET t.embedding = s.embedding; -``` - -**Why:** [Data Evolution](../multimodal-table/data-evolution) mode writes only the updated columns to new files -and merges them at read time. This is ideal for: - -- Adding new feature columns and backfilling data without rewriting the entire table. -- Iterative ML feature engineering — add, update, or refine features as your model evolves. -- Reducing I/O cost and storage overhead for frequent partial column updates. - -#### Distributed Feature Backfill with Ray - -For pipelines that derive features from large payloads, keep the payload in Blob columns and update only the derived -feature columns. Ray can read the records to process, run distributed computation, and call PyPaimon's Ray `merge_into` -to write feature values back to the data-evolution table. The example assumes the target table has a key column `id`, -a Blob column `payload`, and a non-Blob feature column `feature` to update. - -```python -import ray -from pypaimon.ray import read_paimon, merge_into, WhenMatched, source_col - -catalog_options = {"warehouse": "/path/to/warehouse"} -target = "db.item_features" -num_partitions = 1024 - -# Keys selected by an upstream job. This can also come from another Paimon table, -# Parquet files, or any Ray Dataset. -records_to_process = ray.data.read_parquet("/path/to/records-to-process/") - -# Read only the columns needed for this job. -target_rows = read_paimon( - target, - catalog_options=catalog_options, - projection=["id", "payload"], -) - -selected = records_to_process.join( - target_rows, - join_type="inner", - num_partitions=num_partitions, - on=["id"], -) - -def compute_feature(batch): - # Call your model service here. This example only keeps the shape simple. - payloads = batch["payload"].to_pylist() - return { - "id": batch["id"].to_pylist(), - "new_feature": [len(v) if v is not None else 0 for v in payloads], - } - -updates = selected.map_batches(compute_feature, batch_format="pyarrow") - -merge_into( - target=target, - source=updates, - catalog_options=catalog_options, - on=["id"], - when_matched=[ - WhenMatched(update={"feature": source_col("new_feature")}) - ], - num_partitions=num_partitions, -) -``` +### Distributed Feature Backfill with Ray -`merge_into` rewrites only the matched non-Blob columns. Existing Blob files stay unchanged, so the source dataset does -not need to carry Blob columns when it only updates feature fields. For very large key lists, tune Ray resources and -`num_partitions`; selecting target rows is still a distributed join. +The [Ray backfill walkthrough](./ai-pipelines#distributed-feature-backfill-with-ray) shows how to +select records, compute a derived feature, and merge only that feature back into a BLOB table. -### Scenario 11: Python AI Pipeline (PyPaimon) +## Validate Your Choice -**When:** You are building ML training or inference pipelines in Python and need to read/write Paimon tables natively -without JDK dependency. - -**Example: Read data for model training:** - -```python -from pypaimon import CatalogFactory -from torch.utils.data import DataLoader - -# Connect to Paimon -catalog = CatalogFactory.create({'warehouse': 's3://my-bucket/warehouse'}) -table = catalog.get_table('db.feature_store') - -# Read with filter and projection -read_builder = table.new_read_builder() -read_builder = read_builder.with_projection(['user_id', 'embedding']) -read_builder = read_builder.with_filter( - read_builder.new_predicate_builder().equal('gender', 'M') -) -splits = read_builder.new_scan().plan().splits() -table_read = read_builder.new_read() - -# Option 1: Load into PyTorch DataLoader -dataset = table_read.to_torch(splits, streaming=True, prefetch_concurrency=2) -dataloader = DataLoader(dataset, batch_size=32, num_workers=4) -for batch in dataloader: - # Training loop - pass - -# Option 2: Load into Ray for distributed processing -ray_dataset = table_read.to_ray(splits, override_num_blocks=8) -mapped = ray_dataset.map(lambda row: {'feat': row['embedding']}) - -# Option 3: Load into Pandas / PyArrow -df = table_read.to_pandas(splits) -arrow_table = table_read.to_arrow(splits) -``` - -**Why:** [PyPaimon](../pypaimon/) is a pure Python SDK (no JDK required) that integrates seamlessly -with the Python AI ecosystem: - -- **PyTorch**: Direct `DataLoader` integration with streaming and prefetch support. -- **Ray**: Distributed data processing with configurable parallelism. -- **Pandas / PyArrow**: Native DataFrame and Arrow Table support for data science workflows. -- **Data Evolution**: Python API supports `update_by_arrow_with_row_id` and `upsert_by_arrow_with_key` for - row-level updates from Python. See [PyPaimon Data Evolution](../pypaimon/data-evolution). - ---- - -## Summary: Table Type Decision Tree - -``` -Do you need upsert / update / delete? -├── YES → Primary Key Table -│ ├── Simple upsert (keep latest)? → merge-engine = deduplicate (default) -│ ├── Progressive multi-column updates? → merge-engine = partial-update -│ ├── Pre-aggregate metrics? → merge-engine = aggregation -│ └── Dedup keep first? → merge-engine = first-row -│ -│ Table mode: -│ ├── Most scenarios → deletion-vectors.enabled = true (MOW, recommended) -│ ├── Write-heavy, query-light → default MOR -│ └── Read-heavy, batch → full-compaction.delta-commits = 1 (COW) -│ -│ Bucket mode: -│ ├── Most scenarios → bucket = -1 (Dynamic, default) -│ ├── Stable workload, need bucketed join → bucket = N (Fixed) -│ └── Unknown distribution → bucket = -2 (Postpone) -│ -└── NO → Append Table - ├── Standard batch ETL? → No bucket config (unaware-bucket) - │ └── Need query acceleration? → Incremental Clustering or File Index - │ - ├── Need bucket-key filter pushdown / join / ordered streaming? - │ → bucket = N, bucket-key = col (Bucketed Append) - │ - └── AI / Multimodal scenarios? → Enable Data Evolution - ├── Store images / videos / docs? → Blob Table (__BLOB_FIELD comment) - ├── Vector search / RAG? → VECTOR type + Vector Global Index - ├── Feature engineering? → Data Evolution (MERGE INTO partial columns) - └── Python pipeline? → PyPaimon (Ray / PyTorch / Pandas) -``` +Try a representative write and query, including duplicate input, out-of-order updates, and any +required deletes. Verify both the batch result and downstream changelog. Then use +[Understand Files](./understand-files) to inspect the committed layout and +[Streaming Writes and Small Files](./small-files) to assess file growth. diff --git a/docs/docs/learn-paimon/small-files.mdx b/docs/docs/learn-paimon/small-files.mdx new file mode 100644 index 0000000000..4fe5807a3d --- /dev/null +++ b/docs/docs/learn-paimon/small-files.mdx @@ -0,0 +1,161 @@ +--- +title: "Streaming Writes and Small Files" +sidebar_position: 4 +--- + +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Streaming Writes and Small Files + +A small file can be newly written data, a file waiting for compaction, or an old file retained for +history. First identify which kind is growing. Then tune the stage responsible for it. + +## Follow a Streaming Write + +For a typical Flink primary-key sink, the flow is: + + + +1. **Buffer and flush.** Writers buffer records and create data files as buffers fill or when a + checkpoint requires a flush. Files on storage are not queryable merely because they exist. +2. **Prepare a commit.** Writers send pending file changes, including available compaction + results, to the commit path. Flink coordinates publication with successful checkpoint completion. +3. **Publish snapshots.** A commit records the files in manifests and publishes snapshot metadata. + Append and compaction changes can be committed separately; empty input, available compaction + work, and configuration affect the number of snapshots. +4. **Maintain files.** Compaction publishes replacements or metadata changes. Snapshot expiration + later reclaims eligible historical files. With `write-only`, arrange dedicated maintenance. + +Do not estimate checkpoint success from snapshot count alone. Use Flink checkpoint metrics and +the table's `$snapshots` metadata together. For tables using lookup compaction or postpone buckets, +a committed write can also require compaction before the data is visible to the intended reader; +see [Primary-Key Compaction](../primary-key-table/compaction). + +## Diagnose Before Tuning + +Inspect the current snapshot using Flink SQL, replacing `my_table` with your table name: + +```sql +SELECT `partition`, bucket, + COUNT(*) AS file_count, + SUM(file_size_in_bytes) AS total_bytes, + AVG(file_size_in_bytes) AS average_file_bytes +FROM `my_table$files` +GROUP BY `partition`, bucket; + +SELECT snapshot_id, commit_kind, commit_time +FROM `my_table$snapshots` +ORDER BY snapshot_id DESC; +``` + +`$files` describes data files referenced by the selected snapshot. A filesystem or object-store +listing also includes metadata, changelogs, retained history, and possibly orphan files. Compare +like-for-like counts. See [System Tables](../concepts/system-tables) for manifests, consumers, and tags. + +| Observation | Investigate | First action | +| --- | --- | --- | +| Many small live files in every active bucket | Checkpoint interval, input per bucket, and buffer pressure | Measure bytes arriving per checkpoint before changing settings. | +| Increasing overlapping runs in a few buckets | Compaction progress, skew, CPU, memory, and I/O | Restore compaction throughput or rebalance the workload. | +| Few live files, many old data files on storage | Snapshot retention, tags, consumer progress, and expiration execution | Check which references protect the files. | +| Small changelog files dominate | Changelog producer and checkpoint/bucket fan-out | Evaluate `precommit-compact` for changelog consolidation. | +| Inactive partitions retain many live files | Whether maintenance still selects those partitions | Use a dedicated batch compaction for the intended scope. | +| Unreferenced files remain after failed jobs | Uncommitted writer output | Use the documented orphan cleanup with an appropriate age threshold. | + +## Control File Creation + +### Checkpoint Interval + +Short intervals improve commit freshness but leave less time to accumulate data in each active +partition and bucket. Estimate the incoming volume per interval, then account for compression, +record merging, and multiple writers or flushes. + + + +A longer interval can increase file sizes, but also increases visibility latency and changes +recovery behavior. Measure checkpoint duration and your freshness requirement before changing it. +Increasing `target-file-size` alone cannot make a low-volume checkpoint produce a full-sized file. + +### Writer Memory and Spill + +For primary-key writes, `write-buffer-size` influences how much data can be buffered before +flushing or spilling. More memory can reduce premature flushes, but the task must have memory +for its other operators and compaction too. + +`write-buffer-spillable` defaults to `true`. Primary-key write buffers spill locally when an +`IOManager` is available; Flink supplies one. Without it, the buffer remains in memory. Spilling +adds local disk I/O and does not remove checkpoint flushing. See +[Write Performance](../maintenance/write-performance) for the relevant writer options. + +### Partitions and Buckets + +Every active partition and bucket divides the available input. Overly fine partitions or too +many fixed buckets can create small files even when the total table is large. Empty buckets do +not necessarily create files; focus on the destinations receiving records. + +Choose partition keys for pruning and lifecycle needs, then size buckets for the actual +per-partition workload. See [Data Distribution](../primary-key-table/data-distribution) and +[Rescale Bucket](../maintenance/rescale-bucket) before changing an existing layout. + +## Control the Live File Set + +### Primary-Key Tables + +A sorted run contains one or more files. `num-sorted-run.compaction-trigger` is a trigger for +compaction, not a minimum file count per bucket. Runs and files are different units, and both +counts vary as writers and compaction make progress. + +Inspect overlapping runs, file sizes, and compaction duration. Triggering compaction more often +can reduce read-side merging but increase write amplification. See +[Primary-Key Compaction](../primary-key-table/compaction) and +[Sorted Runs](../primary-key-table/#sorted-runs). + +### Append Tables + +Append compaction combines small files. In an ordered bucketed append table, it must preserve +append order within each partition and bucket; it cannot freely merge arbitrary files across +that layout. See [Dedicated Compaction](../maintenance/dedicated-compaction) and +[Bucketed Streaming](../append-table/bucketed#bucketed-streaming). + +Use [Dedicated Compaction](../maintenance/dedicated-compaction) for cold partitions or when +writer-side compaction is disabled. `full-compaction.delta-commits` applies to the Flink writer's +compaction cycle; it is not a background scheduler that revisits every idle partition. Check +compatibility with the table's changelog producer before adding it. + +## Reclaim Historical Files + +Compaction reduces the live file set; it can temporarily increase physical storage because old +snapshots still need the inputs. Shortening retention does not reduce files that remain live. + +Check the required time-travel and recovery window, snapshot count bounds, tags, tracked consumer +progress, and expiration limits. For cleanup procedures, use +[Snapshot Expiration](../maintenance/manage-snapshots#expire-snapshots) and +[Orphan Files](../maintenance/manage-snapshots#remove-orphan-files). Do not infer that a file is orphaned +solely because it is absent from the latest `$files` result. + +Small files add file-open, request, scheduling, and metadata overhead. On HDFS they also increase +NameNode metadata pressure; they do not each consume a full block's worth of data storage merely +because the configured block size is larger than the file. + +## Verify the Improvement + +Compare the same workload before and after the change: live file counts and size distribution, +compaction backlog, checkpoint duration, reader freshness, and query latency. Track physical +storage separately over the retention window. A reduction in one count is useful only if the +required read and write behavior is still met. diff --git a/docs/docs/learn-paimon/understand-files.mdx b/docs/docs/learn-paimon/understand-files.mdx index 5d0a81296a..f94b1e95cb 100644 --- a/docs/docs/learn-paimon/understand-files.mdx +++ b/docs/docs/learn-paimon/understand-files.mdx @@ -1,12 +1,8 @@ --- title: "Understand Files" -sidebar_position: 1 +sidebar_position: 3 --- -import Tabs from '@theme/Tabs'; -import TabItem from '@theme/TabItem'; -import Label from '@site/src/components/Label'; - <!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file @@ -28,483 +24,229 @@ under the License. # Understand Files -This article is specifically designed to clarify -the impact that various file operations have on files. - -This page provides concrete examples and practical tips for -effectively managing them. Furthermore, through an in-depth -exploration of operations such as commit and compact, -we aim to offer insights into the creation and updates of files. +Follow a small primary-key table from creation through inserts, an update, a delete, compaction, +and snapshot expiration. The key distinction is between **logical rows**, **files referenced by +a snapshot**, and **files still present on storage**. ## Prerequisite -Before delving further into this page, please ensure that you have read through the -following sections: +Use a local Flink SQL environment from the [Flink quick start](../flink/quick-start), with a Paimon +catalog and support for [procedures](../flink/procedures). The local filesystem paths below are +for a single-machine experiment. A cluster needs a shared warehouse accessible to its tasks. +Read [Basic Concepts](../concepts/basic-concepts) first if snapshots or manifests are new to you. + +Run the statements in order on a fresh table. Use batch mode and wait for each DML job to finish: -1. [Basic Concepts](../concepts/basic-concepts), -2. [Primary Key Table](../primary-key-table/) and [Append Table](../append-table/) -3. How to use Paimon in [Flink](../flink). +```sql +SET 'execution.runtime-mode' = 'batch'; +SET 'table.dml-sync' = 'true'; +``` ## Understand File Operations ### Create Catalog -Start Flink SQL client via `./sql-client.sh` and execute the following -statements one by one to create a Paimon catalog. ```sql CREATE CATALOG paimon WITH ( -'type' = 'paimon', -'warehouse' = 'file:///tmp/paimon' + 'type' = 'paimon', + 'warehouse' = 'file:///tmp/paimon' ); - USE CATALOG paimon; +USE `default`; ``` -This will only create a directory at given path `file:///tmp/paimon`. +The warehouse is the root for databases and tables. Creating a catalog does not publish a table +data snapshot. ### Create Table -Execute the following create table statement will create a Paimon table with 3 fields: - ```sql -CREATE TABLE T ( - id BIGINT, - a INT, - b STRING, - dt STRING COMMENT 'timestamp string in format yyyyMMdd', - PRIMARY KEY(id, dt) NOT ENFORCED -) PARTITIONED BY (dt); +CREATE TABLE file_demo ( + id BIGINT, + amount INT, + status STRING, + dt STRING, + PRIMARY KEY (id, dt) NOT ENFORCED +) PARTITIONED BY (dt) WITH ( + 'bucket' = '1', + 'file.format' = 'parquet' +); ``` -This will create Paimon table `T` under the path `/tmp/paimon/default.db/T`, -with its schema stored in `/tmp/paimon/default.db/T/schema/schema-0` - +This table has four columns and uses the default `deduplicate` merge engine in merge-on-read +mode. One fixed bucket makes the layout easier to inspect; it is not a production sizing choice. +The initial schema is stored at `/tmp/paimon/default.db/file_demo/schema/schema-0`. ### Insert Records Into Table -Run the following insert statement in Flink SQL: - ```sql -INSERT INTO T VALUES (1, 10001, 'varchar00001', '20230501'); -``` +INSERT INTO file_demo VALUES + (1, 10, 'open', '20260911'), + (2, 20, 'open', '20260911'); -Once the Flink job is completed, the records are written to the Paimon table through a successful `commit`. -Users can verify the visibility of these records by executing the query `SELECT * FROM T` which will return a single row. -The commit process creates a snapshot located at the path `/tmp/paimon/default.db/T/snapshot/snapshot-1`. -The resulting file layout at snapshot-1 is as described below: - - - -The content of snapshot-1 contains metadata of the snapshot, such as manifest list and schema id: -```json -{ - "version" : 3, - "id" : 1, - "schemaId" : 0, - "baseManifestList" : "manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-0", - "deltaManifestList" : "manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-1", - "changelogManifestList" : null, - "commitUser" : "7d758485-981d-4b1a-a0c6-d34c3eb254bf", - "commitIdentifier" : 9223372036854775807, - "commitKind" : "APPEND", - "timeMillis" : 1684155393354, - "logOffsets" : { }, - "totalRecordCount" : 1, - "deltaRecordCount" : 1, - "changelogRecordCount" : 0, - "watermark" : -9223372036854775808 -} +SELECT * FROM file_demo ORDER BY id; +-- (1, 10, 'open', '20260911') +-- (2, 20, 'open', '20260911') ``` -Remind that a manifest list contains all changes of the snapshot, `baseManifestList` is the base -file upon which the changes in `deltaManifestList` is applied. -The first commit will result in 1 manifest file, and 2 manifest lists are -created (the file names might differ from those in your experiment): +The writer creates data files, and a successful commit publishes a snapshot referencing them. +Readers resolve the committed file set through metadata rather than listing every file under +the table directory. -```bash -./T/manifest: -manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-1 -manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-0 -manifest-2b833ea4-d7dc-4de0-ae0d-ad76eced75cc-0 -``` -`manifest-2b833ea4-d7dc-4de0-ae0d-ad76eced75cc-0` is the manifest -file (manifest-1-0 in the above graph), which stores the information about the data files in the snapshot. + -`manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-0` is the -baseManifestList (manifest-list-1-base in the above graph), which is effectively empty. +A simplified directory tree after a write is: -`manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-1` is the -deltaManifestList (manifest-list-1-delta in the above graph), which -contains a list of manifest entries that perform operations on data -files, which, in this case, is `manifest-1-0`. +```text +/tmp/paimon/default.db/file_demo/ +├── schema/ +│ └── schema-0 +├── snapshot/ +│ └── snapshot-<id> +├── manifest/ +│ ├── manifest-list-<base> +│ ├── manifest-list-<delta> +│ └── manifest-<uuid> +└── dt=20260911/ + └── bucket-0/ + └── data-<uuid>.parquet +``` +Names here are symbolic. Snapshot hints and additional metadata can also appear. Flushes and +compaction can change the number of files and snapshots, so do not expect an exact directory +listing or a fixed one-statement-to-one-snapshot mapping. -Now let's insert a batch of records across different partitions and -see what happens. In Flink SQL, execute the following statement: +#### Read the Metadata -```sql -INSERT INTO T VALUES -(2, 10002, 'varchar00002', '20230502'), -(3, 10003, 'varchar00003', '20230503'), -(4, 10004, 'varchar00004', '20230504'), -(5, 10005, 'varchar00005', '20230505'), -(6, 10006, 'varchar00006', '20230506'), -(7, 10007, 'varchar00007', '20230507'), -(8, 10008, 'varchar00008', '20230508'), -(9, 10009, 'varchar00009', '20230509'), -(10, 10010, 'varchar00010', '20230510'); -``` +| Metadata | What it describes | +| --- | --- | +| Schema | Field IDs, types, primary and partition keys, and table options | +| Snapshot | A committed version, including schema ID, commit kind, and manifest references | +| Base manifest list | Manifests describing the base file set for the commit | +| Delta manifest list | Manifests describing the commit's file additions and removals | +| Manifest | File-level `ADD` / `DELETE` entries and metadata such as partition, bucket, and statistics | +| Data file | Physical records; a primary-key file can contain row versions and deletion records | -The second `commit` takes place and executing `SELECT * FROM T` will return -10 rows. A new snapshot, namely `snapshot-2`, is created and gives us the -following physical file layout: -```bash - % ls -1tR . -./T: -dt=20230501 -dt=20230502 -dt=20230503 -dt=20230504 -dt=20230505 -dt=20230506 -dt=20230507 -dt=20230508 -dt=20230509 -dt=20230510 -snapshot -schema -manifest - -./T/snapshot: -LATEST -snapshot-2 -EARLIEST -snapshot-1 - -./T/manifest: -manifest-list-9ac2-5e79-4978-a3bc-86c25f1a303f-1 # delta manifest list for snapshot-2 -manifest-list-9ac2-5e79-4978-a3bc-86c25f1a303f-0 # base manifest list for snapshot-2 -manifest-f1267033-e246-4470-a54c-5c27fdbdd074-0 # manifest file for snapshot-2 - -manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-1 # delta manifest list for snapshot-1 -manifest-list-4ccc-c07f-4090-958c-cfe3ce3889e5-0 # base manifest list for snapshot-1 -manifest-2b833ea4-d7dc-4de0-ae0d-ad76eced75cc-0 # manifest file for snapshot-1 - -./T/dt=20230501/bucket-0: -data-b75b7381-7c8b-430f-b7e5-a204cb65843c-0.orc - -... -# each partition has the data written to bucket-0 -... - -./T/schema: -schema-0 -``` -The new file layout as of snapshot-2 looks like - +The base and delta lists together resolve a snapshot's live data files. A manifest list references +**manifest files**; it does not directly contain the data-file entries. Optional changelog and +index metadata serve separate purposes. See the [Snapshot specification](../concepts/spec/snapshot). -### Delete Records From Table +Inspect the actual commits and live files with Flink SQL: -Now let's delete records that meet the condition `dt>=20230503`. -In Flink SQL, execute the following statement: - -<Label>Batch</Label> ```sql -DELETE FROM T WHERE dt >= '20230503'; -``` -The third `commit` takes place and it gives us `snapshot-3`. Now, listing the files -under the table and your will find out no partition is dropped. Instead, a new data -file is created for partition `20230503` to `20230510`: - -```bash -./T/dt=20230510/bucket-0: -data-b93f468c-b56f-4a93-adc4-b250b3aa3462-0.orc # newer data file created by the delete statement -data-0fcacc70-a0cb-4976-8c88-73e92769a762-0.orc # older data file created by the insert statement -``` +SELECT snapshot_id, schema_id, commit_kind, commit_time +FROM `file_demo$snapshots` +ORDER BY snapshot_id; -This make sense since we insert a record in the second commit (represented by -`+I[10, 10010, 'varchar00010', '20230510']`) and then delete -the record in the third commit. Executing `SELECT * FROM T` will return 2 rows, namely: +SELECT `partition`, bucket, file_path, level, record_count, file_size_in_bytes +FROM `file_demo$files`; ``` -+I[1, 10001, 'varchar00001', '20230501'] -+I[2, 10002, 'varchar00002', '20230502'] -``` - -The new file layout as of snapshot-3 looks like - -Note that `manifest-3-0` contains 8 manifest entries of `ADD` operation type, -corresponding to 8 newly written data files. +`record_count` counts physical records, which can include multiple versions of a key. Use a query +on `file_demo` when you need logical rows. Keep the snapshot IDs from your run for comparisons. +### Update a Record +Insert a new value for the same primary key: -### Compact Table - -As you may have noticed, the number of small files will augment over successive -snapshots, which may lead to decreased read performance. Therefore, a full-compaction -is needed in order to reduce the number of small files. - -Let's trigger the full-compaction now, and run a dedicated compaction job through `flink run`: - -<Label>Batch</Label> - -<Tabs groupId="compact"> - -<TabItem value="flink-sql" label="Flink SQL"> - -```sql -CALL sys.compact( - `table` => 'database_name.table_name', - partitions => 'partition_name', - order_strategy => 'order_strategy', - order_by => 'order_by', - options => 'paimon_table_dynamic_conf' -); -``` +```sql +INSERT INTO file_demo VALUES (1, 15, 'paid', '20260911'); -</TabItem> - -<TabItem value="flink-action" label="Flink Action"> - -```bash -<FLINK_HOME>/bin/flink run \ - -D execution.runtime-mode=batch \ - /path/to/paimon-flink-action-@@VERSION@@.jar \ - compact \ - --warehouse <warehouse-path> \ - --database <database-name> \ - --table <table-name> \ - [--partition <partition-name>] \ - [--catalog_conf <paimon-catalog-conf> [--catalog_conf <paimon-catalog-conf> ...]] \ - [--table_conf <paimon-table-dynamic-conf> [--table_conf <paimon-table-dynamic-conf>] ...] +SELECT * FROM file_demo ORDER BY id; +-- (1, 15, 'paid', '20260911') +-- (2, 20, 'open', '20260911') ``` -</TabItem> - -</Tabs> +A new record supersedes the old version of key `(1, '20260911')`. The reader merges versions to +produce the latest row. This does not require overwriting the original data file in place. -an example would be (suppose you're already in Flink home) - -<Tabs groupId="compact-example"> +### Delete Records From Table -<TabItem value="flink-sql" label="Flink SQL"> +Use a predicate on a non-partition column to exercise row-level deletion: ```sql -CALL sys.compact('T'); -``` +DELETE FROM file_demo WHERE id = 2; -</TabItem> - -<TabItem value="flink-action" label="Flink Action"> - -```bash -./bin/flink run \ - ./lib/paimon-flink-action-@@VERSION@@.jar \ - compact \ - --path file:///tmp/paimon/default.db/T +SELECT * FROM file_demo ORDER BY id; +-- (1, 15, 'paid', '20260911') ``` -</TabItem> - -</Tabs> - -All current table files will be compacted and a new snapshot, namely `snapshot-4`, is -made and contains the following information: - -```json -{ - "version" : 3, - "id" : 4, - "schemaId" : 0, - "baseManifestList" : "manifest-list-9be16-82e7-4941-8b0a-7ce1c1d0fa6d-0", - "deltaManifestList" : "manifest-list-9be16-82e7-4941-8b0a-7ce1c1d0fa6d-1", - "changelogManifestList" : null, - "commitUser" : "a3d951d5-aa0e-4071-a5d4-4c72a4233d48", - "commitIdentifier" : 9223372036854775807, - "commitKind" : "COMPACT", - "timeMillis" : 1684163217960, - "logOffsets" : { }, - "totalRecordCount" : 2, - "deltaRecordCount" : -16, - "changelogRecordCount" : 0, - "watermark" : -9223372036854775808 -} -``` +In this MOR example, a deletion record cancels the earlier row when versions are merged. +A new file containing deletion records is represented by a manifest **`ADD`** entry. A manifest +**`DELETE`** entry instead retires an entire data file from a snapshot. These are different kinds +of deletion. A partition-only SQL delete can use a different path and is not the example here. -The new file layout as of snapshot-4 looks like - +### Compact Table -Note that `manifest-4-0` contains 20 manifest entries (18 `DELETE` operations and 2 `ADD` operations) -1. For partition `20230503` to `20230510`, two `DELETE` operations for two data files -2. For partition `20230501` to `20230502`, one `DELETE` operation and one `ADD` operation for the same data file. - This is because there has been an upgrade of the file from level 0 to the highest level. Please rest assured that - this is only a change in metadata, and the file is still the same. +In the same batch SQL session: -### Alter Table -Execute the following statement to configure full-compaction: ```sql -ALTER TABLE T SET ('full-compaction.delta-commits' = '1'); +CALL sys.compact('default.file_demo'); ``` -It will create a new schema for Paimon table, namely `schema-1`, but no snapshot -has actually used this schema yet until the next commit. - -### Expire Snapshots - -Remind that the marked data files are not truly deleted until the snapshot expires and -no consumer depends on the snapshot. For more information, see [Expiring Snapshots](../maintenance/manage-snapshots#expire-snapshots). +Full compaction resolves row versions in the selected buckets and commits the resulting file +changes. The query still returns `(1, 15, 'paid', '20260911')`. Inspect `$snapshots` and `$files` +again to see which files changed. Compaction can replace files or promote a file's level using +metadata; it need not rewrite every file. See [Compaction](../primary-key-table/compaction). -During the process of snapshot expiration, the range of snapshots is initially determined, and then data files within these snapshots are marked for deletion. -A data file is `marked` for deletion only when there is a manifest entry of kind `DELETE` that references that specific data file. -This marking ensures that the file will not be utilized by subsequent snapshots and can be safely removed. + +The diagram groups updates and deletes into one stage. Letters label illustrative files, not +actual filenames or a promised file count. Older snapshots can still reference the original files +after compaction has removed those files from the latest snapshot. -Let's say all 4 snapshots in the above diagram are about to expire. The expire process is as follows: - -1. It first deletes all marked data files, and records any changed buckets. - -2. It then deletes any changelog files and associated manifests. - -3. Finally, it deletes the snapshots themselves and writes the earliest hint file. - -If any directories are left empty after the deletion process, they will be deleted as well, -but only when `snapshot.clean-empty-directories` is enabled (default is `false`). -By default, empty directories are kept on disk. See [Manage Snapshots](../maintenance/manage-snapshots#expire-snapshots). - - -Let's say another snapshot, `snapshot-5` is created and snapshot expiration is triggered. `snapshot-1` to `snapshot-4` are -to be deleted. For simplicity, we will only focus on files from previous snapshots, the final layout after snapshot -expiration looks like: - - - -As a result, partition `20230503` to `20230510` are physically deleted. - -### Flink Stream Write - -Finally, we will examine Flink Stream Write by utilizing the example -of CDC ingestion. This section will address the capturing and writing of -change data into Paimon, as well as the mechanisms behind asynchronous compact -and snapshot commit and expiration. - -To begin, let's take a closer look at the CDC ingestion workflow and -the unique roles played by each component involved. - - - -1. `MySQL CDC Source` uniformly reads snapshot and incremental data, with `SnapshotReader` reading snapshot data - and `BinlogReader` reading incremental data, respectively. -2. `Paimon Sink` writes data into Paimon table in bucket level. The `CompactManager` within it will trigger compaction - asynchronously. -3. `Committer Operator` is a singleton responsible for committing and expiring snapshots. - -Next, we will go over end-to-end data flow. - - - - -`MySQL Cdc Source` read snapshot and incremental data and emit them to downstream after normalization. - - - - -`Paimon Sink` first buffers new records in a heap-based LSM tree, and flushes them to disk when -the memory buffer is full. Note that each data file written is a sorted run. At this point, no manifest file and snapshot -is created. Right before Flink checkpoint takes places, `Paimon Sink` will flush all buffered records and send committable message -to downstream, which is read and committed by `Committer Operator` during checkpoint. - - - -During checkpoint, `Committer Operator` will create a new snapshot and associate it with manifest lists so that the snapshot -contains information about all data files in the table. - - - -At later point asynchronous compaction might take place, and the committable produced by `CompactManager` contains information -about previous files and merged files so that `Committer Operator` can construct corresponding manifest entries. In this case -`Committer Operator` might produce two snapshot during Flink checkpoint, one for data written (snapshot of kind `Append`) and the -other for compact (snapshot of kind `Compact`). If no data file is written during checkpoint interval, only snapshot of kind `Compact` -will be created. `Committer Operator` will check against snapshot expiration and perform -physical deletion of marked data files. - -## Understand Small Files - -Many users are concerned about small files, which can lead to: -1. Stability issue: Too many small files in HDFS, NameNode will be overstressed. -2. Cost issue: A small file in HDFS will temporarily use the size of a minimum of one Block, for example 128 MB. -3. Query efficiency: The efficiency of querying too many small files will be affected. - -### Understand Checkpoints - -Assuming you are using Flink Writer, each checkpoint generates 1-2 snapshots, and the checkpoint forces the files to be -generated on DFS, so the smaller the checkpoint interval the more small files will be generated. - -1. So first thing is increase checkpoint interval. - -By default, not only checkpoint will cause the file to be generated, but writer's memory (write-buffer-size) exhaustion -will also flush data to DFS and generate the corresponding file. You can enable `write-buffer-spillable` to generate -spilled files in writer to generate bigger files in DFS. - -2. So second thing is increase `write-buffer-size` or enable `write-buffer-spillable`. - -### Understand Snapshots - - - -Paimon maintains multiple versions of files, compaction and deletion of files are logical and do not actually -delete files. Files are only really deleted when Snapshot is expired, so the first way to reduce files is to -reduce the time it takes for snapshot to be expired. Flink writer will automatically expire snapshots. - -See [Expire Snapshots](../maintenance/manage-snapshots#expire-snapshots). - -### Understand Partitions and Buckets - -Paimon files are organized in a layered style. The following image illustrates the file layout. Starting -from a snapshot file, Paimon readers can recursively access all records from the table. - - +### Alter Table -For example, the following table: +Change an option without writing data: ```sql -CREATE TABLE MyTable ( - user_id BIGINT, - item_id BIGINT, - behavior STRING, - dt STRING, - hh STRING, - PRIMARY KEY (dt, hh, user_id) NOT ENFORCED -) PARTITIONED BY (dt, hh) WITH ( - 'bucket' = '10' -); +ALTER TABLE file_demo SET ('full-compaction.delta-commits' = '1'); ``` -The table data will be physically sliced into different partitions, and different buckets inside, so if the overall -data volume is too small, there is at least one file in a single bucket, I suggest you configure a smaller number -of buckets, otherwise there will be quite a few small files as well. - -### Understand LSM for Primary Table +This creates a new schema version. Existing snapshots retain their recorded schema IDs; a later +data commit can reference the updated schema. The option requests frequent full compaction for +subsequent writes, with additional write cost. It does not itself publish a new data snapshot or +rewrite existing files. See [Table Mode](../primary-key-table/table-mode#copy-on-write). -LSM tree organizes files into several sorted runs. A sorted run consists of one or multiple data files and each data -file belongs to exactly one sorted run. +### Expire Snapshots - +Snapshot expiration reclaims obsolete files when retention rules and protected references allow +it. A file is not deleted merely because the snapshot that first introduced it has expired: +newer snapshots may still use that same file. -By default, sorted runs number depends on `num-sorted-run.compaction-trigger`, see [Compaction for Primary Key Table](../primary-key-table/compaction), -this means that there are at least 5 files in a bucket. If you want to reduce this number, you can keep fewer files, but write performance may suffer. +| After an operation | Latest query | Physical storage | +| --- | --- | --- | +| Row update or delete | Reflects merged row changes | Can still contain earlier row versions | +| Compaction | Same logical result | Can contain both old and replacement files | +| Expiration of obsolete history | Same logical result | Eligible retired files and expired metadata are reclaimed | -### Understand Files for Bucketed Append Table +Retention combines `snapshot.time-retained`, `snapshot.num-retained.min`, and +`snapshot.num-retained.max`. Tags and tracked consumers can protect history or its files; see +[Manage Snapshots](../maintenance/manage-snapshots#expire-snapshots). The latest required state +must remain available even as older snapshots expire. -By default, Append also does automatic compaction to reduce the number of small files. +After files are reclaimed, empty directories are kept by default. Cleaning them requires +`snapshot.clean-empty-directories = true`, so an empty partition directory alone is not evidence +of failed data cleanup. Files left by unsuccessful writes need the separate +[orphan-file cleanup](../maintenance/manage-snapshots#remove-orphan-files) workflow. -However, for Bucketed Append table, it will only compact the files within the Bucket for sequential -purposes, which may keep more small files. See [Bucketed Append](../append-table/bucketed). +## Flink Stream Write {#flink-stream-write} -### Understand Full-Compaction +Streaming writes repeat the same commit and compaction lifecycle around checkpoints. Continue +with [Streaming Writes and Small Files](./small-files#follow-a-streaming-write) for the operator +flow, visibility, and file-size implications. -Maybe you think the 5 files for the primary key table are actually okay, but the Append table (bucket) -may have 50 small files in a single bucket, which is very difficult to accept. Worse still, partitions that -are no longer active also keep so many small files. +## Understand Small Files -Configure 'full-compaction.delta-commits' perform full-compaction periodically in Flink writing. And it can ensure -that partitions are full compacted before writing ends. +Use the [small-file diagnosis guide](./small-files#diagnose-before-tuning) to determine whether +files belong to the current snapshot, retained history, changelogs, or unsuccessful writes. +Reducing live files and reclaiming old files require different actions. + +| Topic | Continue reading | +| --- | --- | +| <a id="understand-checkpoints"></a>Checkpoints | [Checkpoint interval](./small-files#checkpoint-interval) | +| <a id="understand-snapshots"></a>Snapshots | [Reclaim historical files](./small-files#reclaim-historical-files) | +| <a id="understand-partitions-and-buckets"></a>Distribution | [Partitions and buckets](./small-files#partitions-and-buckets) | +| <a id="understand-lsm-for-primary-table"></a>LSM files | [Primary-key tables](./small-files#primary-key-tables) | +| <a id="understand-files-for-bucketed-append-table"></a>Bucketed append files | [Append tables](./small-files#append-tables) | +| <a id="understand-full-compaction"></a>Full compaction | [Dedicated compaction](../maintenance/dedicated-compaction) | diff --git a/docs/docs/pypaimon/ray-joins.md b/docs/docs/pypaimon/ray-joins.md index f47d819486..0d091973b7 100644 --- a/docs/docs/pypaimon/ray-joins.md +++ b/docs/docs/pypaimon/ray-joins.md @@ -235,7 +235,7 @@ counts the rows actually updated or deleted (after condition filtering). `num_unchanged` is `0` in the current implementation. For an end-to-end feature update workflow on Blob tables, see -[Distributed Feature Backfill with Ray](../learn-paimon/scenario-guide#distributed-feature-backfill-with-ray). +[Distributed Feature Backfill with Ray](../learn-paimon/ai-pipelines#distributed-feature-backfill-with-ray). **Notes:** - Partition key columns cannot be updated by matched update clauses, because diff --git a/docs/sidebars.js b/docs/sidebars.js index 788cb2c615..eb32e6c79f 100644 --- a/docs/sidebars.js +++ b/docs/sidebars.js @@ -687,8 +687,10 @@ const sidebars = { "id": "learn-paimon/index" }, "items": [ + "learn-paimon/scenario-guide", + "learn-paimon/ai-pipelines", "learn-paimon/understand-files", - "learn-paimon/scenario-guide" + "learn-paimon/small-files" ] } ], diff --git a/docs/static/img/learn-paimon-ai-pipeline.svg b/docs/static/img/learn-paimon-ai-pipeline.svg new file mode 100644 index 0000000000..7bac688939 --- /dev/null +++ b/docs/static/img/learn-paimon-ai-pipeline.svg @@ -0,0 +1,69 @@ +<svg xmlns="http://www.w3.org/2000/svg"; viewBox="0 0 1080 736" role="img" aria-labelledby="title desc"> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<title id="title">A pipeline for payloads, features, and search</title> +<desc id="desc">Ingest BLOB payloads and scalar metadata into a Data Evolution table. Read selected columns and compute features, then commit only the updated columns while retaining untouched payload files. Training readers can project the committed columns. Search requires a separate index build or refresh and a coverage policy for newly written rows.</desc> +<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7" markerHeight="7" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="#526277"/></marker></defs> +<rect width="1080" height="736" rx="16" fill="#ffffff"/> +<g font-family="Arial, Helvetica, sans-serif"> +<text x="32" y="48" font-size="28" font-weight="700" text-anchor="start" fill="#172b4d">A pipeline for payloads, features, and search</text> +<text x="32" y="80" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Keep large payloads reusable; track column versions and search coverage separately.</text> +<rect x="32" y="126" width="300" height="130" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="50" y="158" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">1 Ingest</text> +<text x="50" y="188" font-size="17" font-weight="400" text-anchor="start" fill="#526277">BLOB payloads</text> +<text x="50" y="215" font-size="17" font-weight="400" text-anchor="start" fill="#526277">Scalar metadata + application ID</text> +<rect x="390" y="126" width="300" height="130" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="408" y="158" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">2 Compute</text> +<text x="408" y="188" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Project the required columns</text> +<text x="408" y="215" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Run inference in Ray or Spark</text> +<rect x="748" y="126" width="300" height="130" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="766" y="158" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">3 Commit features</text> +<text x="766" y="188" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Update selected columns</text> +<text x="766" y="215" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Preserve untouched payloads</text> +<path d="M 332 191 H 390" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 690 191 H 748" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<rect x="32" y="346" width="658" height="150" rx="10" fill="#f1f4f8" stroke="#64748b" stroke-width="1.5"/> +<text x="50" y="378" font-size="21" font-weight="700" text-anchor="start" fill="#64748b">Data Evolution table</text> +<text x="50" y="408" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Normal columns · dedicated BLOBs · optional vector files</text> +<text x="50" y="435" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Readers align column versions by row ID.</text> +<text x="50" y="462" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Updates preserve the complete affected normal-file range.</text> +<path d="M 182 256 V 346" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 898 256 V 310 H 650 V 346" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<rect x="748" y="346" width="300" height="150" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="766" y="378" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">Training / analysis</text> +<text x="766" y="408" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Select snapshot or tag</text> +<text x="766" y="435" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Read projected columns</text> +<text x="766" y="462" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Use PyTorch, Arrow, or Ray</text> +<path d="M 690 421 H 748" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<rect x="32" y="568" width="300" height="116" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="50" y="600" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">4 Build / refresh index</text> +<text x="50" y="630" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Explicit indexing operation</text> +<text x="50" y="657" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Choose coverage policy</text> +<rect x="390" y="568" width="300" height="116" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="408" y="600" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">5 Serve search</text> +<text x="408" y="630" font-size="18" font-weight="400" text-anchor="start" fill="#526277">ANN / scalar / text queries</text> +<text x="408" y="657" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Validate freshness and recall</text> +<path d="M 182 496 V 568" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 332 626 H 390" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="748" y="584" font-size="18" font-weight="400" text-anchor="start" fill="#526277">New writes do not automatically</text> +<text x="748" y="612" font-size="18" font-weight="400" text-anchor="start" fill="#526277">refresh an existing index.</text> +<text x="748" y="640" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Fast search can omit rows</text> +<text x="748" y="668" font-size="18" font-weight="400" text-anchor="start" fill="#526277">outside index coverage.</text> +</g> +</svg> diff --git a/docs/static/img/learn-paimon-file-lifecycle.svg b/docs/static/img/learn-paimon-file-lifecycle.svg new file mode 100644 index 0000000000..e0cbff75b0 --- /dev/null +++ b/docs/static/img/learn-paimon-file-lifecycle.svg @@ -0,0 +1,65 @@ +<svg xmlns="http://www.w3.org/2000/svg"; viewBox="0 0 1080 792" role="img" aria-labelledby="title desc"> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<title id="title">Row changes, file changes, and physical cleanup</title> +<desc id="desc">In a merge-on-read primary-key table, A holds rows 1 and 2. New files record an update to row 1 and deletion of row 2. Compaction replaces the active files with C containing row 1 paid. Old input files can remain for history until expiration can safely delete them. The latest file C remains.</desc> +<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7" markerHeight="7" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="#526277"/></marker></defs> +<rect width="1080" height="792" rx="16" fill="#ffffff"/> +<g font-family="Arial, Helvetica, sans-serif"> +<text x="32" y="48" font-size="28" font-weight="700" text-anchor="start" fill="#172b4d">Row changes, file changes, and physical cleanup</text> +<text x="32" y="80" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Illustrative MOR lifecycle; file letters and counts are schematic.</text> +<text x="32" y="122" font-size="18" font-weight="700" text-anchor="start" fill="#172b4d">Operation</text> +<text x="238" y="122" font-size="18" font-weight="700" text-anchor="start" fill="#172b4d">Latest snapshot reads</text> +<text x="675" y="122" font-size="18" font-weight="700" text-anchor="start" fill="#172b4d">Query result</text> +<text x="875" y="122" font-size="18" font-weight="700" text-anchor="start" fill="#172b4d">Old inputs</text> +<text x="32" y="188" font-size="21" font-weight="700" text-anchor="start" fill="#172b4d">1 Append</text> +<rect x="220" y="148" width="430" height="104" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="238" y="180" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">A: (1, open), (2, open)</text> +<text x="238" y="210" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Two current rows</text> +<text x="675" y="191" font-size="20" font-weight="700" text-anchor="start" fill="#10705d">1 open, 2 open</text> +<text x="875" y="188" font-size="18" font-weight="400" text-anchor="start" fill="#526277">None</text> +<path d="M 435 252 V 290" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="32" y="330" font-size="21" font-weight="700" text-anchor="start" fill="#172b4d">2 Update / delete</text> +<rect x="220" y="290" width="430" height="104" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="238" y="322" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">A + new change files</text> +<text x="238" y="352" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Update 1 → paid; delete 2</text> +<text x="675" y="333" font-size="20" font-weight="700" text-anchor="start" fill="#10705d">1 paid</text> +<text x="875" y="330" font-size="18" font-weight="400" text-anchor="start" fill="#526277">May still be live</text> +<text x="875" y="359" font-size="16" font-weight="400" text-anchor="start" fill="#526277">before compaction</text> +<path d="M 435 394 V 432" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="32" y="472" font-size="21" font-weight="700" text-anchor="start" fill="#172b4d">3 Compact</text> +<rect x="220" y="432" width="430" height="104" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="238" y="464" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">C: (1, paid)</text> +<text x="238" y="494" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Retire inputs from latest snapshot</text> +<text x="675" y="475" font-size="20" font-weight="700" text-anchor="start" fill="#10705d">1 paid</text> +<text x="875" y="472" font-size="18" font-weight="400" text-anchor="start" fill="#526277">A + changes</text> +<text x="875" y="501" font-size="17" font-weight="400" text-anchor="start" fill="#946200">kept for history</text> +<path d="M 435 536 V 574" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="32" y="614" font-size="21" font-weight="700" text-anchor="start" fill="#172b4d">4 Expire history</text> +<rect x="220" y="574" width="430" height="104" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="238" y="606" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">C: (1, paid)</text> +<text x="238" y="636" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Keep files required by latest state</text> +<text x="675" y="617" font-size="20" font-weight="700" text-anchor="start" fill="#10705d">1 paid</text> +<text x="875" y="614" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Eligible inputs</text> +<text x="875" y="643" font-size="17" font-weight="400" text-anchor="start" fill="#10705d">reclaimed</text> +<path d="M 32 706 L 1048 706" stroke="#d5dee9" stroke-width="1.5"/> +<text x="32" y="740" font-size="20" font-weight="700" text-anchor="start" fill="#172b4d">Row DELETE ≠ manifest DELETE</text> +<text x="32" y="770" font-size="18" font-weight="400" text-anchor="start" fill="#526277">A file containing row deletions is added; a manifest DELETE removes a file reference from a snapshot.</text> +</g> +</svg> diff --git a/docs/static/img/learn-paimon-small-files.svg b/docs/static/img/learn-paimon-small-files.svg new file mode 100644 index 0000000000..ea7d51e29e --- /dev/null +++ b/docs/static/img/learn-paimon-small-files.svg @@ -0,0 +1,61 @@ +<svg xmlns="http://www.w3.org/2000/svg"; viewBox="0 0 1080 656" role="img" aria-labelledby="title desc"> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<title id="title">Why a large stream can produce small files</title> +<desc id="desc">The same illustrative 80 megabytes of input per checkpoint spreads into 40 megabytes for each of two active partition-bucket destinations, or 10 megabytes for each of eight active destinations. More destinations or shorter intervals reduce data per destination. This is input volume before compression, merging, or extra flushes, not a prediction of output file size.</desc> +<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7" markerHeight="7" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="#526277"/></marker></defs> +<rect width="1080" height="656" rx="16" fill="#ffffff"/> +<g font-family="Arial, Helvetica, sans-serif"> +<text x="32" y="48" font-size="28" font-weight="700" text-anchor="start" fill="#172b4d">Why a large stream can produce small files</text> +<text x="32" y="80" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Illustrative input volume per checkpoint, before compression, merging, or extra flushes.</text> +<rect x="314" y="112" width="452" height="82" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="332" y="144" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">80 MB arriving in one checkpoint</text> +<path d="M 540 194 V 224 H 278 V 274" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 540 224 H 802 V 274" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<rect x="32" y="274" width="492" height="268" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="50" y="306" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">2 active partition / bucket destinations</text> +<rect x="556" y="274" width="492" height="268" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="574" y="306" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">8 active partition / bucket destinations</text> +<rect x="66" y="346" width="202" height="124" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="84" y="378" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">40 MB</text> +<text x="84" y="408" font-size="18" font-weight="400" text-anchor="start" fill="#526277">per destination</text> +<rect x="290" y="346" width="202" height="124" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="308" y="378" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">40 MB</text> +<text x="308" y="408" font-size="18" font-weight="400" text-anchor="start" fill="#526277">per destination</text> +<rect x="580" y="340" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="631" y="380" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="696" y="340" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="747" y="380" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="812" y="340" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="863" y="380" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="928" y="340" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="979" y="380" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="580" y="428" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="631" y="468" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="696" y="428" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="747" y="468" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="812" y="428" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="863" y="468" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<rect x="928" y="428" width="102" height="66" rx="8" fill="#ffffff" stroke="#2166b1" stroke-width="1.5"/> +<text x="979" y="468" font-size="19" font-weight="700" text-anchor="middle" fill="#2166b1">10 MB</text> +<text x="578" y="526" font-size="18" font-weight="400" text-anchor="start" fill="#526277">More destinations → less data in each</text> +<text x="32" y="589" font-size="21" font-weight="700" text-anchor="start" fill="#172b4d">Other causes to check</text> +<text x="32" y="622" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Short intervals · buffer pressure · skew · compaction backlog · retained history</text> +</g> +</svg> diff --git a/docs/static/img/learn-paimon-streaming-commit.svg b/docs/static/img/learn-paimon-streaming-commit.svg new file mode 100644 index 0000000000..3d031f38d3 --- /dev/null +++ b/docs/static/img/learn-paimon-streaming-commit.svg @@ -0,0 +1,66 @@ +<svg xmlns="http://www.w3.org/2000/svg"; viewBox="0 0 1080 676" role="img" aria-labelledby="title desc"> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<title id="title">From Flink records to a visible snapshot</title> +<desc id="desc">Writers buffer input, flush data files and prepare committables. The commit path publishes manifests and snapshots in coordination with successful checkpoint completion. Compaction also supplies file changes to the commit path. Snapshot expiration reclaims eligible old files later. A file existing on storage alone does not make it readable.</desc> +<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7" markerHeight="7" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="#526277"/></marker></defs> +<rect width="1080" height="676" rx="16" fill="#ffffff"/> +<g font-family="Arial, Helvetica, sans-serif"> +<text x="32" y="48" font-size="28" font-weight="700" text-anchor="start" fill="#172b4d">From Flink records to a visible snapshot</text> +<text x="32" y="80" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Publication follows checkpoint success; file creation alone does not publish table data.</text> +<rect x="32" y="126" width="300" height="116" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="50" y="158" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">1 Receive and buffer</text> +<text x="50" y="188" font-size="19" font-weight="400" text-anchor="start" fill="#526277">CDC / stream records</text> +<text x="50" y="215" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Writer memory and local spill</text> +<rect x="390" y="126" width="300" height="116" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="408" y="158" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">2 Flush and prepare</text> +<text x="408" y="188" font-size="17" font-weight="400" text-anchor="start" fill="#526277">Write pending data files</text> +<text x="408" y="215" font-size="17" font-weight="400" text-anchor="start" fill="#526277">Prepare file-change committables</text> +<rect x="748" y="126" width="300" height="116" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="766" y="158" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">3 Publish commit</text> +<text x="766" y="188" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Manifests + snapshot</text> +<text x="766" y="215" font-size="19" font-weight="400" text-anchor="start" fill="#526277">After checkpoint success</text> +<path d="M 332 184 H 390" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 690 184 H 748" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="407" y="274" font-size="17" font-weight="400" text-anchor="start" fill="#946200">Pending files are not yet visible.</text> +<rect x="390" y="342" width="300" height="116" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="408" y="374" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">Compaction</text> +<text x="408" y="404" font-size="17" font-weight="400" text-anchor="start" fill="#526277">Merge or promote files</text> +<text x="408" y="431" font-size="17" font-weight="400" text-anchor="start" fill="#526277">Produce replacement file changes</text> +<path d="M 540 342 V 302 H 716 V 224 H 748" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="733" y="307" font-size="17" font-weight="400" text-anchor="start" fill="#526277">Same commit path</text> +<rect x="748" y="342" width="300" height="116" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="766" y="374" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">Readers</text> +<text x="766" y="404" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Select committed snapshot</text> +<text x="766" y="431" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Apply table-mode read rules</text> +<path d="M 898 242 V 342" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<rect x="748" y="530" width="300" height="106" rx="10" fill="#f1f4f8" stroke="#64748b" stroke-width="1.5"/> +<text x="766" y="562" font-size="21" font-weight="700" text-anchor="start" fill="#64748b">Snapshot expiration</text> +<text x="766" y="592" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Reclaim eligible old files</text> +<path d="M 1048 222 H 1062 V 583 H 1048" fill="none" stroke="#526277" stroke-width="2" stroke-dasharray="6 5" marker-end="url(#arrow)"/> +<text x="32" y="360" font-size="21" font-weight="700" text-anchor="start" fill="#172b4d">Two different maintenance jobs</text> +<text x="32" y="400" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Compaction reduces live-file work.</text> +<text x="32" y="430" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Expiration reclaims retained history.</text> +<text x="32" y="460" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Neither is guaranteed by a fixed</text> +<text x="32" y="490" font-size="18" font-weight="400" text-anchor="start" fill="#526277">snapshot count per checkpoint.</text> +<text x="32" y="562" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Lookup and postpone layouts can add</text> +<text x="32" y="590" font-size="18" font-weight="400" text-anchor="start" fill="#526277">a compaction dependency for visibility.</text> +<text x="32" y="618" font-size="18" font-weight="400" text-anchor="start" fill="#526277">See the table-mode requirements.</text> +</g> +</svg> diff --git a/docs/static/img/learn-paimon-table-choice.svg b/docs/static/img/learn-paimon-table-choice.svg new file mode 100644 index 0000000000..15f84e0c8e --- /dev/null +++ b/docs/static/img/learn-paimon-table-choice.svg @@ -0,0 +1,55 @@ +<svg xmlns="http://www.w3.org/2000/svg"; viewBox="0 0 1080 680" role="img" aria-labelledby="title desc"> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<title id="title">Choose row semantics before storage options</title> +<desc id="desc">Incoming rows either merge by primary key or remain independent append rows. Primary-key designs then choose distribution, read mode, and changelog; append designs choose unaware buckets, fixed buckets, or Data Evolution for column updates and managed payloads.</desc> +<defs><marker id="arrow" viewBox="0 0 10 10" refX="9" refY="5" markerWidth="7" markerHeight="7" orient="auto-start-reverse"><path d="M 0 0 L 10 5 L 0 10 z" fill="#526277"/></marker></defs> +<rect width="1080" height="680" rx="16" fill="#ffffff"/> +<g font-family="Arial, Helvetica, sans-serif"> +<text x="32" y="48" font-size="28" font-weight="700" text-anchor="start" fill="#172b4d">Choose row semantics before storage options</text> +<text x="32" y="80" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Start with the meaning of an input record, then check each feature’s requirements.</text> +<rect x="230" y="112" width="620" height="78" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="248" y="144" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">Should incoming records merge by a declared key?</text> +<path d="M 540 190 V 214 H 278 V 256" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 540 214 H 802 V 256" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<text x="292" y="242" font-size="18" font-weight="700" text-anchor="start" fill="#2166b1">Yes</text> +<text x="816" y="242" font-size="18" font-weight="700" text-anchor="start" fill="#10705d">No</text> +<rect x="32" y="256" width="492" height="124" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="50" y="288" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">Primary-key table</text> +<text x="50" y="318" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Keep latest · combine columns · aggregate</text> +<text x="50" y="345" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Or keep the first row received for a key</text> +<rect x="556" y="256" width="492" height="124" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="574" y="288" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">Append table</text> +<text x="574" y="318" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Store rows independently</text> +<text x="574" y="345" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Batch SQL can still update or delete rows</text> +<path d="M 278 380 V 426" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<path d="M 802 380 V 426" fill="none" stroke="#526277" stroke-width="2" marker-end="url(#arrow)"/> +<rect x="32" y="426" width="492" height="178" rx="10" fill="#edf4ff" stroke="#2166b1" stroke-width="1.5"/> +<text x="50" y="458" font-size="21" font-weight="700" text-anchor="start" fill="#2166b1">Make three separate choices</text> +<text x="50" y="488" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Distribution: dynamic / fixed / postpone</text> +<text x="50" y="515" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Read behavior: MOR / COW / MOW</text> +<text x="50" y="542" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Streaming output: changelog producer</text> +<rect x="556" y="426" width="492" height="178" rx="10" fill="#eaf7f2" stroke="#10705d" stroke-width="1.5"/> +<text x="574" y="458" font-size="21" font-weight="700" text-anchor="start" fill="#10705d">Choose a layout for the workload</text> +<text x="574" y="488" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Unaware buckets: batch and general append</text> +<text x="574" y="515" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Fixed buckets: pruning / joins / append order</text> +<text x="574" y="542" font-size="19" font-weight="400" text-anchor="start" fill="#526277">Data Evolution: column updates and BLOBs</text> +<text x="32" y="646" font-size="18" font-weight="400" text-anchor="start" fill="#526277">Indexes are an additional choice. An embedding column does not automatically have a search index.</text> +</g> +</svg>
