Merge remote-tracking branch 'upstream/master' into disk_ann

* upstream/master: (33 commits)
  [core] Fix merge adjacent files in DataEvolutionCompactCoordinator
  [python] Rename list_tag to list_tags
  [python] add list tag for TagManager (apache#7264)
  [core][python] Introduce DataFileMeta.nonNullRowIdRange to unify codes
  [python] with_shard should be evenly distributed for data evolution mode (apache#7271)
  [core] Remove useless version in Varant
  [core] Should work with Split in DataTableBatchScan
  [core] Fix  paimon_incremental_query with limit push down (apache#7269)
  [rest] Improve RestCatalog OpenAPI nonce generation (apache#7270)
  [cdc] Avoid sending empty schema change events to Schema Evolution (apache#7261)
  [python] Fix avro write timestamp without timezone wrongly (apache#7259)
  [doc] add doc for filter by _ROW_ID on data evolution (apache#7262)
  [fs] Extract jindo dls to separate module (apache#7263)
  [core] Add listTableDetails method to Catalog interface (apache#7266)
  [python] Support filter by _ROW_ID for data evolution (apache#7252)
  [Flink] Add sourceParallelismUpperBound metric for auto-scaling systems (apache#7117)
  [github] Add whether it is an AI-generated tag in the PR template (apache#7257)
  [core] Improve HttpClient error response handling (apache#7254)
  [python] Light refactor: move _is_blob_file check into DataFileMeta (apache#7256)
  [core] RowIdPredicateVisitor supports converting between statement (apache#7255)
  ...

Author: jerry-024

.github/PULL_REQUEST_TEMPLATE.md

@@ -18,3 +18,12 @@ Linked issue: close #xxx
 ### Documentation
 
 <!-- Does this change introduce a new feature -->
+
+### Generative AI tooling
+
+<!--
+If generative AI tooling has been used in the process of authoring this patch, please include the
+phrase: 'Generated-by: ' followed by the name of the tool and its version.
+If no, write 'No'.
+Please refer to the [ASF Generative Tooling Guidance](https://www.apache.org/legal/generative-tooling.html) for details.
+-->

docs/content/append-table/blob.md

@@ -122,6 +122,24 @@ For details about the blob file format structure, see [File Format - BLOB]({{< r
 
 *Required for blob functionality to work correctly.
 
+Specifically, if the storage system of the input BlobDescriptor differs from that used by Paimon, 
+you can specify the storage configuration for the input blob descriptor using the prefix 
+`blob-descriptor.`. For example, if the source data is stored in a different OSS endpoint, 
+you can configure it as below (using flink sql as an example):
+```sql
+CREATE TABLE image_table (
+    id INT,
+    name STRING,
+    image BYTES
+) WITH (
+    'row-tracking.enabled' = 'true',
+    'data-evolution.enabled' = 'true',
+    'blob-field' = 'image',
+    'fs.oss.endpoint' = 'aaa',                   -- This is for Paimon's own config
+    'blob-descriptor.fs.oss.endpoint' = 'bbb'    -- This is for input blob descriptors' config
+);
+```
+
 ## SQL Usage
 
 ### Creating a Table

docs/content/maintenance/metrics.md

@@ -391,6 +391,12 @@ When using Flink to read and write, Paimon has implemented some key standard Fli
             <td>Gauge</td>
             <td>Time difference between reading the data file and file creation.</td>
         </tr>
+        <tr>
+            <td>sourceParallelismUpperBound</td>
+            <td>Flink Source Enumerator</td>
+            <td>Gauge</td>
+            <td>Recommended upper bound of parallelism for auto-scaling systems. Note: This is a recommendation, not a hard limit.</td>
+        </tr>
     </tbody>
 </table>
 

docs/content/pypaimon/data-evolution.md

@@ -98,6 +98,16 @@ table_commit.close()
 #   'f1': [-1001, 1002]
 ```
 
+## Filter by _ROW_ID
+
+Requires the same [Prerequisites](#prerequisites) (row-tracking and data-evolution enabled). On such tables you can filter by `_ROW_ID` to prune files at scan time. Supported: `equal('_ROW_ID', id)`, `is_in('_ROW_ID', [id1, ...])`, `between('_ROW_ID', low, high)`.
+
+```python
+pb = table.new_read_builder().new_predicate_builder()
+rb = table.new_read_builder().with_filter(pb.equal('_ROW_ID', 0))
+result = rb.new_read().to_arrow(rb.new_scan().plan().splits())
+```
+
 ## Update Columns By Shards
 
 If you want to **compute a derived column** (or **update an existing column based on other columns**) without providing

docs/content/pypaimon/manage-tags.md

@@ -0,0 +1,61 @@
+---
+title: "Manage Tags"
+weight: 3
+type: docs
+aliases:
+  - /pypaimon/manage-tags.html
+---
+
+<!--
+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.
+-->
+
+# Manage Tags
+
+Just like Java API of Paimon, you can create a [tag]({{< ref "maintenance/manage-tags" >}}) based on a snapshot. The tag will maintain the manifests and data files of the snapshot. 
+A typical usage is creating tags daily, then you can maintain the historical data of each day for batch reading.
+## Create and Delete Tag
+
+You can create a tag with given name and snapshot ID, and delete a tag with given name.
+
+```python
+
+table = catalog.get_table('database_name.table_name')
+table.create_tag("tag2", snapshot_id=2)  # create tag2 based on snapshot 2
+table.create_tag("tag2")  # create tag2 based on latest snapshot
+table.delete_tag("tag2")  # delete tag2
+```
+
+If snapshot_id unset, snapshot_id defaults to the latest.
+
+## Read Tag
+You can read data from a specific tag.
+```python
+
+table = catalog.get_table('database_name.table_name')
+table.create_tag("tag2", snapshot_id=2)
+
+# Read from tag2 using scan.tag-name option
+table_with_tag = table.copy({"scan.tag-name": "tag2"})
+read_builder = table_with_tag.new_read_builder()
+table_scan = read_builder.new_scan()
+table_read = read_builder.new_read()
+result = table_read.to_arrow(table_scan.plan().splits())
+```
+
+

docs/content/pypaimon/python-api.md

@@ -258,7 +258,7 @@ predicate_5 = predicate_builder.and_predicates([predicate3, predicate4])
 read_builder = read_builder.with_filter(predicate_5)
 ```
 
-See [Predicate]({{< ref "python-api#predicate" >}}) for all supported filters and building methods.
+See [Predicate]({{< ref "python-api#predicate" >}}) for all supported filters and building methods. Filter by `_ROW_ID`: see [Data Evolution]({{< ref "pypaimon/data-evolution#filter-by-_row_id" >}}).
 
 You can also pushdown projection by `ReadBuilder`:
 

docs/layouts/shortcodes/generated/core_configuration.html

@@ -1452,6 +1452,24 @@
             <td>String</td>
             <td>The Variant shredding schema for writing.</td>
         </tr>
+        <tr>
+            <td><h5>visibility-callback.check-interval</h5></td>
+            <td style="word-wrap: break-word;">10 s</td>
+            <td>Duration</td>
+            <td>The interval for checking visibility when visibility-callback enabled.</td>
+        </tr>
+        <tr>
+            <td><h5>visibility-callback.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to enable the visibility wait callback that waits for compaction to complete after commit. This is useful for primary key tables with deletion vectors or postpone bucket mode to ensure data visibility, only used for batch mode or bounded stream.</td>
+        </tr>
+        <tr>
+            <td><h5>visibility-callback.timeout</h5></td>
+            <td style="word-wrap: break-word;">30 min</td>
+            <td>Duration</td>
+            <td>The maximum time to wait for compaction to complete when visibility callback is enabled. If the timeout is reached, an exception will be thrown.</td>
+        </tr>
         <tr>
             <td><h5>write-buffer-for-append</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -90,6 +90,8 @@ public class CoreOptions implements Serializable {
 
     public static final String COLUMNS = "columns";
 
+    public static final String BLOB_DESCRIPTOR_PREFIX = "blob-descriptor.";
+
     public static final ConfigOption<TableType> TYPE =
             key("type")
                     .enumType(TableType.class)
@@ -2204,6 +2206,30 @@ public InlineElement getDescription() {
                     .withDescription(
                             "Whether to try upgrading the data files after overwriting a primary key table.");
 
+    public static final ConfigOption<Boolean> VISIBILITY_CALLBACK_ENABLED =
+            key("visibility-callback.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to enable the visibility wait callback that waits for compaction to complete "
+                                    + "after commit. This is useful for primary key tables with deletion vectors or "
+                                    + "postpone bucket mode to ensure data visibility, only used for batch mode or bounded stream.");
+
+    public static final ConfigOption<Duration> VISIBILITY_CALLBACK_TIMEOUT =
+            key("visibility-callback.timeout")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(30))
+                    .withDescription(
+                            "The maximum time to wait for compaction to complete when visibility callback is enabled. "
+                                    + "If the timeout is reached, an exception will be thrown.");
+
+    public static final ConfigOption<Duration> VISIBILITY_CALLBACK_CHECK_INTERVAL =
+            key("visibility-callback.check-interval")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(10))
+                    .withDescription(
+                            "The interval for checking visibility when visibility-callback enabled.");
+
     private final Options options;
 
     public CoreOptions(Map<String, String> options) {
@@ -3431,6 +3457,18 @@ public boolean overwriteUpgrade() {
         return options.get(OVERWRITE_UPGRADE);
     }
 
+    public boolean visibilityCallbackEnabled() {
+        return options.get(VISIBILITY_CALLBACK_ENABLED);
+    }
+
+    public Duration visibilityCallbackTimeout() {
+        return options.get(VISIBILITY_CALLBACK_TIMEOUT);
+    }
+
+    public Duration visibilityCallbackCheckInterval() {
+        return options.get(VISIBILITY_CALLBACK_CHECK_INTERVAL);
+    }
+
     /** Specifies the merge engine for table with primary key. */
     public enum MergeEngine implements DescribedEnum {
         DEDUPLICATE("deduplicate", "De-duplicate and keep the last row."),

paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java

@@ -131,19 +131,14 @@ private <T extends RESTResponse> T exec(HttpUriRequestBase request, Class<T> res
                     response -> {
                         String responseBodyStr = RESTUtil.extractResponseBodyAsString(response);
                         if (!RESTUtil.isSuccessful(response)) {
-                            ErrorResponse error;
+                            ErrorResponse error = null;
                             try {
                                 error = RESTApi.fromJson(responseBodyStr, ErrorResponse.class);
                             } catch (JsonProcessingException e) {
-                                error =
-                                        new ErrorResponse(
-                                                null,
-                                                null,
-                                                responseBodyStr != null
-                                                        ? responseBodyStr
-                                                        : "response body is null",
-                                                response.getCode());
+                                // ignore exception
                             }
+                            error = buildErrorResponse(error, responseBodyStr, response.getCode());
+
                             errorHandler.accept(error, extractRequestId(response));
                         }
                         if (responseType != null && responseBodyStr != null) {
@@ -228,4 +223,22 @@ private static Header[] getHeaders(
                 .map(entry -> new BasicHeader(entry.getKey(), entry.getValue()))
                 .toArray(Header[]::new);
     }
+
+    private static ErrorResponse buildErrorResponse(
+            ErrorResponse error, String responseBodyStr, int errorCode) {
+        if (error == null || error.getMessage() == null || error.getMessage().isEmpty()) {
+            String resourceType =
+                    (error != null && error.getResourceType() != null)
+                            ? error.getResourceType()
+                            : "";
+            String resourceName =
+                    (error != null && error.getResourceName() != null)
+                            ? error.getResourceName()
+                            : "";
+            String message = responseBodyStr != null ? responseBodyStr : "response body is null";
+            int code = (error != null && error.getCode() != null) ? error.getCode() : errorCode;
+            error = new ErrorResponse(resourceType, resourceName, message, code);
+        }
+        return error;
+    }
 }

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -420,6 +420,28 @@ public PagedList<GetTableResponse> listTableDetailsPaged(
         return new PagedList<>(tables, response.getNextPageToken());
     }
 
+    /**
+     * List table details for a database.
+     *
+     * <p>Gets an array of table details for a database. There is no guarantee of a specific
+     * ordering of the elements in the array.
+     *
+     * @param databaseName name of database.
+     * @return a list of table details.
+     * @throws NoSuchResourceException Exception thrown on HTTP 404 means the database not exists
+     * @throws ForbiddenException Exception thrown on HTTP 403 means don't have the permission for
+     *     this database
+     */
+    public List<GetTableResponse> listTableDetails(String databaseName) {
+        return listDataFromPageApi(
+                queryParams ->
+                        client.get(
+                                resourcePaths.tableDetails(databaseName),
+                                queryParams,
+                                ListTableDetailsResponse.class,
+                                restAuthFunction));
+    }
+
     /**
      * List table for a catalog.
      *