merge with master

Author: mikedias

AGENTS.md

@@ -21,8 +21,8 @@ Prefer running the narrowest tests.
 
 #### Java Tests (surefire)
 
-- Single method: `mvn -pl <module> -Dtest=TestClassName#methodName test`
-- Single class: `mvn -pl <module> -Dtest=TestClassName test`
+- Single method: `mvn -pl <module> -DwildcardSuites=none -Dtest=TestClassName#methodName test`
+- Single class: `mvn -pl <module> -DwildcardSuites=none -Dtest=TestClassName test`
 
 #### Scala Tests (scalatest)
 
@@ -34,6 +34,18 @@ mvn -pl paimon-spark/paimon-spark-ut -am -Pfast-build -DfailIfNoTests=false \
   -Dtest=none test
 ```
 
+#### Required Maven Profiles
+- Spark tests: add `-Pspark3` or `-Pspark4`
+- Flink tests: add `-Pflink1` or `-Pflink2`
+
+For example, run a Java test in the Spark UT with Spark 3:
+```shell
+mvn -pl paimon-spark/paimon-spark-ut -am -Pfast-build -Pspark3 \
+  -DwildcardSuites=none \
+  -DfailIfNoTests=false \
+  -Dtest=SparkChainTableITCase#testChainTable test
+```
+
 ### Local Iteration Speedup
 
 Use `-Pfast-build` to skip checkstyle, spotless, enforcer, and rat checks (not for final verification):

LICENSE

@@ -272,6 +272,9 @@ paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java
 paimon-format/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java
 from https://parquet.apache.org/ version 1.14.0
 
+paimon-format/src/main/java/org/apache/paimon/format/ArrowSchemaMetadata.java
+from https://arrow.apache.org/ version 15.0.0
+
 paimon-common/src/main/java/org/apache/paimon/data/variant/GenericVariant.java
 paimon-common/src/main/java/org/apache/paimon/data/variant/GenericVariantBuilder.java
 paimon-common/src/main/java/org/apache/paimon/data/variant/GenericVariantUtil.java

docs/docs/concepts/spec/fileformat.md

@@ -849,13 +849,33 @@ BLOB files use the `.blob` extension and have the following structure:
 +------------------+
 ```
 
+Each physical BLOB file stores one logical field. The field can be either `BLOB` or
+`ARRAY<BLOB>`. For `ARRAY<BLOB>`, the variable-length data area in an entry uses the
+following nested payload:
+
+```
++----------------------+-----------------------------------------------+
+| Array Magic Number   | 4 bytes (1094861634, Little Endian)           |
+| Array Version        | 1 byte                                        |
+| Element Count        | 4 bytes (Little Endian)                       |
+| Element Data         | Concatenated bytes of all non-null elements   |
+| Element Length Index | Delta-Varint compressed element lengths       |
+| Index Length         | 4 bytes (Little Endian)                       |
++----------------------+-----------------------------------------------+
+```
+
+An element length of `-1` represents a null array element. At the outer file index
+level, `-1` represents a null field and `-2` represents a field placeholder used by
+data evolution. An empty array is encoded with an element count of zero and an empty
+element index; it is distinct from a null array.
+
 Key features:
 - **CRC32 Checksums**: Each blob entry has a CRC32 checksum for data integrity verification
 - **Indexed Access**: The index at the end enables efficient random access to any blob in the file
 - **Delta-Varint Compression**: The index uses delta-varint compression for space efficiency
 
 Limitations:
-1. BLOB format only supports a single BLOB type column per file.
+1. BLOB format only supports a single `BLOB` or `ARRAY<BLOB>` field per physical file.
 2. BLOB format does not support predicate pushdown.
 3. Statistics collection is not supported for BLOB columns.
 

docs/docs/concepts/system-tables.mdx

@@ -439,16 +439,17 @@ You can query the partition files of the table.
 SELECT * FROM my_table$partitions;
 
 /*
-+-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+
-| partition | record_count | file_size_in_bytes| file_count | last_update_time    | created_at          | created_by | updated_by | options | total_buckets | done  |
-+-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+
-| {1}       |            1 |               645 |          1 | 2024-06-24 10:25:57 | 2024-06-24 10:20:00 | admin      | test_user  | {}      |             1 | false |
-+-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+
++-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+----------------------+
+| partition | record_count | file_size_in_bytes| file_count | last_update_time    | created_at          | created_by | updated_by | options | total_buckets | done  | deleted_record_count |
++-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+----------------------+
+| {1}       |            1 |               645 |          1 | 2024-06-24 10:25:57 | 2024-06-24 10:20:00 | admin      | test_user  | {}      |             1 | false |                    0 |
++-----------+--------------+-------------------+------------+---------------------+---------------------+------------+------------+---------+---------------+-------+----------------------+
 */
 ```
 
 **Note**: 
 - The `created_by`, `created_at`, `updated_by`, and `options` fields are populated from REST catalog audit information. For non-REST catalogs, these fields will be `NULL`.
+- The `deleted_record_count` field is the number of records marked deleted by deletion vectors in the partition. It is `0` when the partition has no deletion vectors and `NULL` when legacy deletion vector metadata does not contain cardinality.
 
 ### Buckets Table
 

docs/docs/flink/lineage.md

@@ -0,0 +1,38 @@
+---
+title: "Data Lineage"
+sidebar_position: 13
+---
+
+# Data Lineage
+
+Paimon implements [FLIP-314](https://cwiki.apache.org/confluence/spaces/FLINK/pages/255070913/FLIP-314+Support+Customized+Job+Lineage+Listener) to expose lineage information through Flink's native [LineageVertexProvider](https://nightlies.apache.org/flink/flink-docs-stable/docs/internals/data_lineage/) interface. This allows lineage consumers (such as [OpenLineage](https://openlineage.io/docs/integrations/flink/flink2)) to automatically discover Paimon datasets in the Flink job graph.
+
+:::info
+Data Lineage support is available only for Flink 2.0 and above.
+:::
+
+## Table API / SQL
+
+When using Paimon tables via Flink SQL or Table API, lineage is automatically reported for both source and sink tables.
+
+Apart from the table lineage information provided by Flink natively (name, schema), Paimon enriches each dataset with a namespace and a **config facet** containing core table options and Iceberg metadata options.
+
+:::warning
+Full Table API lineage support requires <a href="https://github.com/apache/flink/pull/27727" style="color: #1677ff; text-decoration: underline;">this upstream Flink change</a> which is pending merge. Without it, Flink will not call `getLineageVertex()` for `DataStreamScanProvider` and `DataStreamSinkProvider`, which can lead to potentially missing Paimon input or output datasets. This applies to all Flink 2.x versions that do not yet include this change.
+:::
+
+## DataStream API
+
+Paimon also supports lineage for the DataStream API. Lineage is automatically reported for sources built via `FlinkSourceBuilder` and for all sinks that extend `FlinkSink`.
+
+Similar to the Table API, it reports all lineage information including name, namespace, schema facet, and config facet.
+
+## Configuration
+
+To consume lineage events, a [JobStatusChangedListener](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/advanced/job_status_listener/#configuration) must be added in your Flink configuration.
+
+For example, to configure with OpenLineage:
+
+```
+execution.job-status-changed-listeners = io.openlineage.flink.listener.OpenLineageJobStatusChangedListenerFactory
+```

docs/docs/flink/procedures.md

@@ -37,7 +37,7 @@ argument can be omitted. For the above example, the call statement is \
 ``CALL sys.compact(`table` => 'default.t', options => 'sink.parallelism=4')``.
 
 Specify partitions: we use string to represent partition filter. "," means "AND" and ";" means "OR". For example, if you want
-to specify two partitions date=01 and date=02, you need to write 'date=01;date=02'; If you want to specify one partition
+to specify two partitions date=01 or date=02, you need to write 'date=01;date=02'; If you want to specify one partition
 with date=01 and day=01, you need to write 'date=01,day=01'.
 
 Table options syntax: we use string to represent table options. The format is 'key1=value1,key2=value2...'.
@@ -120,8 +120,8 @@ All available procedures are listed below.
       <td>
          To compact databases. Arguments:
             <li>includingDatabases: to specify databases. You can use regular expression.</li>
-            <li>mode: compact mode. "divided": start a sink for each table, detecting the new table requires restarting the job;
-               "combined" (default): start a single combined sink for all tables, the new table will be automatically detected.
+            <li>mode: compact mode. "divided" (default): start a sink for each table, detecting the new table requires restarting the job;
+               "combined": start a single combined sink for all tables, the new table will be automatically detected.
             </li>
             <li>includingTables: to specify tables. You can use regular expression.</li>
             <li>excludingTables: to specify tables that are not compacted. You can use regular expression.</li>
@@ -139,6 +139,29 @@ All available procedures are listed below.
             compat_strategy => 'full')
       </td>
    </tr>
+   <tr>
+      <td>compact_chain_table</td>
+      <td>
+         -- Use named argument<br/>
+         CALL [catalog.]sys.compact_chain_table(
+            `table` => 'table',
+            partition => 'partition',
+            overwrite => overwrite) <br/><br/>
+         -- Use indexed argument<br/>
+         CALL [catalog.]sys.compact_chain_table('table', 'partition') <br/>
+         CALL [catalog.]sys.compact_chain_table('table', 'partition', overwrite) <br/><br/>
+      </td>
+      <td>
+         To compact chain table by merging snapshot and delta branches into the snapshot branch. Arguments:
+            <li>table: the target chain table identifier. Cannot be empty.</li>
+            <li>partition: partition specification format (e.g., 'dt=20250810,hour=22'). Cannot be empty.</li>
+            <li>overwrite: whether to overwrite if the partition already exists in the snapshot branch. Default is false. Optional.</li>
+      </td>
+      <td>
+         CALL sys.compact_chain_table(`table` => 'default.T', partition => 'dt=20250810,hour=22')<br/><br/>
+         CALL sys.compact_chain_table('default.T', 'dt=20250810,hour=22', true)
+      </td>
+   </tr>
    <tr>
       <td>create_tag</td>
       <td>
@@ -1117,6 +1140,35 @@ All available procedures are listed below.
             `dry_run` => true)
       </td>
    </tr>
+   <tr>
+      <td>full_text_search</td>
+      <td>
+         CALL [catalog.]sys.full_text_search(<br/>
+            `table` => 'identifier',<br/>
+            `column` => 'columnName',<br/>
+            query => 'queryJson',<br/>
+            top_k => topK,<br/>
+            projection => 'col1,col2,__paimon_search_score',<br/>
+            options => 'key1=value1;key2=value2')<br/>
+      </td>
+      <td>
+         To perform full-text search on a table and return deterministically ordered JSON rows. Arguments:
+            <li>table(required): the target table identifier.</li>
+            <li>column(required): the character column to search.</li>
+            <li>query(required): native full-text query JSON.</li>
+            <li>top_k(required): the maximum number of results, from 1 through 10,000.</li>
+            <li>projection(optional): comma-separated result columns. Add <code>__paimon_search_score</code> to return the search relevance score.</li>
+            <li>options(optional): additional dynamic options of the table. The query authorization setting cannot be overridden.</li>
+      </td>
+      <td>
+         CALL sys.full_text_search(<br/>
+            `table` => 'default.articles',<br/>
+            `column` => 'content',<br/>
+            query => '{"match":{"column":"content","terms":"paimon lake"}}',<br/>
+            top_k => 10,<br/>
+            projection => 'id,content,__paimon_search_score')
+      </td>
+   </tr>
    <tr>
       <td>vector_search</td>
       <td>

docs/docs/maintenance/metrics.md

@@ -37,7 +37,7 @@ Paimon has supported built-in metrics to measure operations of **commits**, **sc
 
 ## Metrics List
 
-Below is lists of Paimon built-in metrics. They are summarized into types of scan metrics, commit metrics, write metrics, write buffer metrics and compaction metrics.
+Below is lists of Paimon built-in metrics. They are summarized into types of scan metrics, commit metrics, write metrics, write buffer metrics, blob fetch metrics and compaction metrics.
 
 ### Scan Metrics
 
@@ -226,6 +226,85 @@ Below is lists of Paimon built-in metrics. They are summarized into types of sca
     </tbody>
 </table>
 
+### Blob Fetch Metrics
+
+<table class="table table-bordered">
+    <thead>
+    <tr>
+      <th class="text-left" style="width: 225pt">Metrics Name</th>
+      <th class="text-left" style="width: 70pt">Type</th>
+      <th class="text-left" style="width: 300pt">Description</th>
+    </tr>
+    </thead>
+    <tbody>
+        <tr>
+            <td>blobFetchTotal</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch outcomes handled during write, including successful fetches, failures, and missing resources written as NULL.</td>
+        </tr>
+        <tr>
+            <td>blobFetchSuccess</td>
+            <td>Counter</td>
+            <td>Total number of blob fetches that completed successfully.</td>
+        </tr>
+        <tr>
+            <td>blobFetchSuccessBytes</td>
+            <td>Counter</td>
+            <td>Total bytes written by successful blob fetches.</td>
+        </tr>
+        <tr>
+            <td>blobFetchNullWritten</td>
+            <td>Counter</td>
+            <td>Total number of NULL values written because a blob resource was missing or failed to fetch.</td>
+        </tr>
+        <tr>
+            <td>blobFetchMissingFileNullWritten</td>
+            <td>Counter</td>
+            <td>Total number of NULL values written because <code>blob-write-null-on-missing-file</code> handled a missing blob resource.</td>
+        </tr>
+        <tr>
+            <td>blobFetchFailureNullWritten</td>
+            <td>Counter</td>
+            <td>Total number of NULL values written because <code>blob-write-null-on-fetch-failure</code> handled a non-404 blob fetch failure.</td>
+        </tr>
+        <tr>
+            <td>blobFetchFailure</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures that were not written as NULL.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpNotFound</td>
+            <td>Counter</td>
+            <td>Total number of HTTP 404 blob fetch failures.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpClientError</td>
+            <td>Counter</td>
+            <td>Total number of non-404 HTTP 4xx blob fetch failures.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpServerError</td>
+            <td>Counter</td>
+            <td>Total number of HTTP 5xx blob fetch failures.</td>
+        </tr>
+        <tr>
+            <td>blobFetchHttpOtherError</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures with HTTP status codes outside 4xx and 5xx.</td>
+        </tr>
+        <tr>
+            <td>blobFetchInvalidUri</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures caused by invalid URIs.</td>
+        </tr>
+        <tr>
+            <td>blobFetchOtherError</td>
+            <td>Counter</td>
+            <td>Total number of blob fetch failures that are not classified as HTTP status or invalid URI failures.</td>
+        </tr>
+    </tbody>
+</table>
+
 ### Compaction Metrics
 
 <table class="table table-bordered">
@@ -375,6 +454,11 @@ From Flink Web-UI, go to the committer operator's metrics, it's shown as:
             <td>&lt;host&gt;.taskmanager.&lt;tm_id&gt;.&lt;job_name&gt;.&lt;writer_operator_name&gt;.&lt;subtask_index&gt;</td>
             <td>paimon.table.&lt;table_name&gt;.writeBuffer</td>
         </tr>
+        <tr>
+            <td>Blob Fetch Metrics</td>
+            <td>&lt;host&gt;.taskmanager.&lt;tm_id&gt;.&lt;job_name&gt;.&lt;writer_operator_name&gt;.&lt;subtask_index&gt;</td>
+            <td>paimon.table.&lt;table_name&gt;.blobFetch</td>
+        </tr>
         <tr>
             <td>Compaction Metrics</td>
             <td>&lt;host&gt;.taskmanager.&lt;tm_id&gt;.&lt;job_name&gt;.&lt;writer_operator_name&gt;.&lt;subtask_index&gt;</td>

docs/docs/maintenance/write-performance.md

@@ -126,7 +126,7 @@ execution.checkpointing.timeout = 60 min
 In the initialization of write, the writer of the bucket needs to read all historical files. If there is a bottleneck
 here (For example, writing a large number of partitions simultaneously), you can use `sink.writer-coordinator.enabled`
 to use a Flink coordinator to cache the read manifest data to accelerate initialization. The cache memory for coordinator
-is `sink.writer-coordinator.cache-memory`, default is 1GB in Job Manager.
+is `sink.writer-coordinator.cache-memory`, default is 2GB in Job Manager.
 
 The coordinator manifest cache normally holds entries with soft references, so the JVM can reclaim them when the Job
 Manager runs low on heap. On a heavily loaded Job Manager this can backfire: the JVM reclaims cached manifests, writers