You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/03/26 01:47:33 UTC
(paimon) branch master updated: [core] Chain table support group partition. (#7524)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f261a66bcc [core] Chain table support group partition. (#7524)
f261a66bcc is described below
commit f261a66bcc0b5a5b48e0f0086e0a3d7c85c40855
Author: Wenchao Wu <60...@users.noreply.github.com>
AuthorDate: Thu Mar 26 09:47:28 2026 +0800
[core] Chain table support group partition. (#7524)
[Feature] chain table support group partition. #7522
Previously, the chain table merge logic assumed all partition fields
participate in the chain dimension. For example, a table partitioned by
(region, date) was treated as a single flat sequence, meaning a
chain-table query for region=US could incorrectly pick up a snapshot
anchor from region=EU.
This change introduces the concept of a group partition: the non-chain
prefix fields of the partition key. Delta partitions are now grouped by
their group key before performing the chain-merge, so that the anchor
search for each group is scoped correctly.
---
docs/content/primary-key-table/chain-table.md | 59 +
.../shortcodes/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 17 +
.../org/apache/paimon/schema/SchemaValidation.java | 23 +
.../apache/paimon/table/ChainGroupReadTable.java | 307 +++--
.../paimon/utils/ChainPartitionProjector.java | 160 +++
.../org/apache/paimon/utils/ChainTableUtils.java | 132 ++
.../paimon/utils/ChainPartitionProjectorTest.java | 432 ++++++
.../apache/paimon/utils/ChainTableUtilsTest.java | 532 +++++++
.../apache/paimon/spark/SparkChainTableITCase.java | 1456 ++++++++++++++++++++
10 files changed, 3008 insertions(+), 116 deletions(-)
diff --git a/docs/content/primary-key-table/chain-table.md b/docs/content/primary-key-table/chain-table.md
index d8d4120f91..839eb12fc9 100644
--- a/docs/content/primary-key-table/chain-table.md
+++ b/docs/content/primary-key-table/chain-table.md
@@ -147,3 +147,62 @@ you will get the following result:
| 2 | 1| 1 |
+---+----+-----+
```
+
+## Group Partition
+
+In real-world scenarios, a table often has multiple partition dimensions. For example, data may be
+partitioned by both `region` and `date`. In such cases, different regions are independent data silos —
+each should maintain its own chain independently rather than sharing one global chain across all regions.
+
+Paimon supports this pattern via **group partition**: partition keys are divided into two parts:
+- **Group partition keys** (prefix fields): Dimensions that identify independent data silos (e.g., `region`).
+ Each distinct combination of group partition values forms its own independent chain.
+- **Chain partition keys** (suffix fields): Dimensions that form the time-ordered chain within a group
+ (e.g., `date`).
+
+Use `chain-table.chain-partition-keys` to specify the chain dimension. This value must be a
+**contiguous suffix** of the table's partition keys. Partition fields before it automatically become the
+group dimension. If this option is not set, all partitions belong to a single implicit group (the
+default behavior for single-dimension partitioned tables).
+
+Consider an example where the table is partitioned by `region` and `date`, and you want each region to
+have its own chain:
+
+```sql
+CREATE TABLE default.t (
+ `t1` string ,
+ `t2` string ,
+ `t3` string
+) PARTITIONED BY (`region` string, `date` string)
+TBLPROPERTIES (
+ 'chain-table.enabled' = 'true',
+ 'primary-key' = 'region,date,t1',
+ 'sequence.field' = 't2',
+ 'bucket-key' = 't1',
+ 'bucket' = '2',
+ 'partition.timestamp-pattern' = '$date',
+ 'partition.timestamp-formatter' = 'yyyyMMdd',
+ -- specify that only `date` is the chain dimension; `region` becomes the group dimension
+ 'chain-table.chain-partition-keys' = 'date'
+);
+```
+
+With this configuration:
+- Partition keys: `[region, date]`
+- Group partition keys: `[region]` — CN and US each have their own independent chain
+- Chain partition keys: `[date]` — time-ordered chain within each region
+
+When reading a partition like `(region='CN', date='20250811')`, Paimon finds the nearest earlier
+snapshot partition **within the same region** (e.g., `(region='CN', date='20250810')`) as the chain
+anchor, and merges forward through the delta data for the CN group only. The US group is resolved
+independently using its own anchor.
+
+For hourly partitioned tables with a regional dimension, you can set both `dt` and `hour` as chain
+partition keys:
+
+```sql
+'chain-table.chain-partition-keys' = 'dt,hour'
+```
+
+This treats `(dt, hour)` as the composite chain dimension and everything before it (e.g., `region`) as
+the group dimension.
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 67618e4e83..220ca4cdc4 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -1187,6 +1187,12 @@ This config option does not affect the default filesystem metastore.</td>
<td>String</td>
<td>When a batch job queries from a chain table, if a partition does not exist in the main branch, the reader will try to get this partition from chain snapshot branch.</td>
</tr>
+ <tr>
+ <td><h5>chain-table.chain-partition-keys</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Partition keys that participate in chain logic. Must be a contiguous suffix of the table's partition keys. Comma-separated. If not set, all partition keys participate in chain.</td>
+ </tr>
<tr>
<td><h5>scan.file-creation-time-millis</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index e8ae69e9be..4dc495c12b 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -273,6 +273,15 @@ public class CoreOptions implements Serializable {
"When a batch job queries from a chain table, if a partition does not exist in either main or snapshot branch, "
+ "the reader will try to get this partition from chain snapshot and delta branch together.");
+ public static final ConfigOption<String> CHAIN_TABLE_CHAIN_PARTITION_KEYS =
+ key("chain-table.chain-partition-keys")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Partition keys that participate in chain logic. Must be a contiguous "
+ + "suffix of the table's partition keys. Comma-separated. "
+ + "If not set, all partition keys participate in chain.");
+
public static final String FILE_FORMAT_ORC = "orc";
public static final String FILE_FORMAT_AVRO = "avro";
public static final String FILE_FORMAT_PARQUET = "parquet";
@@ -3694,6 +3703,14 @@ public class CoreOptions implements Serializable {
return options.get(SCAN_FALLBACK_DELTA_BRANCH);
}
+ public List<String> chainTableChainPartitionKeys() {
+ String value = options.get(CHAIN_TABLE_CHAIN_PARTITION_KEYS);
+ if (value == null || value.trim().isEmpty()) {
+ return null;
+ }
+ return Arrays.stream(value.split(",")).map(String::trim).collect(Collectors.toList());
+ }
+
public boolean formatTableImplementationIsPaimon() {
return options.get(FORMAT_TABLE_IMPLEMENTATION) == FormatTableImplementation.PAIMON;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 130f9b29eb..b9b68fc647 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -784,6 +784,29 @@ public class SchemaValidation {
Preconditions.checkArgument(
options.partitionTimestampFormatter() != null,
"Partition timestamp formatter is required for chain table.");
+
+ // validate chain-table.chain-partition-keys
+ List<String> chainPartKeys = options.chainTableChainPartitionKeys();
+ if (chainPartKeys != null) {
+ List<String> partitionKeys = schema.partitionKeys();
+
+ Preconditions.checkArgument(
+ !chainPartKeys.isEmpty(),
+ "chain-table.chain-partition-keys must not be empty.");
+
+ // chain partition keys should be in the tails of partition keys.
+ int chainStart = partitionKeys.size() - chainPartKeys.size();
+ Preconditions.checkArgument(
+ chainStart >= 0
+ && partitionKeys
+ .subList(chainStart, partitionKeys.size())
+ .equals(chainPartKeys),
+ "chain-table.chain-partition-keys must be a contiguous suffix "
+ + "of the table's partition keys. "
+ + "Partition keys: %s, chain partition keys: %s.",
+ partitionKeys,
+ chainPartKeys);
+ }
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
index 2379882070..78752acfc6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java
@@ -39,9 +39,9 @@ import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.ChainPartitionProjector;
import org.apache.paimon.utils.ChainTableUtils;
import org.apache.paimon.utils.Filter;
-import org.apache.paimon.utils.InternalRowPartitionComputer;
import org.apache.paimon.utils.Pair;
import org.apache.paimon.utils.RowDataToObjectArrayConverter;
@@ -129,10 +129,11 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable {
public static class ChainTableBatchScan extends FallbackReadScan {
private final RowDataToObjectArrayConverter partitionConverter;
- private final InternalRowPartitionComputer partitionComputer;
private final CoreOptions options;
private final RecordComparator partitionComparator;
private final ChainGroupReadTable chainGroupReadTable;
+ private final RecordComparator chainPartitionComparator;
+ private final ChainPartitionProjector partitionProjector;
private Predicate dataPredicate;
private Filter<Integer> bucketFilter;
@@ -151,21 +152,28 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable {
this.chainGroupReadTable = chainGroupReadTable;
this.partitionConverter =
new RowDataToObjectArrayConverter(tableSchema.logicalPartitionType());
- this.partitionComputer =
- new InternalRowPartitionComputer(
- options.partitionDefaultName(),
- tableSchema.logicalPartitionType(),
- tableSchema.partitionKeys().toArray(new String[0]),
- options.legacyPartitionName());
this.partitionComparator =
CodeGenUtils.newRecordComparator(
tableSchema.logicalPartitionType().getFieldTypes());
+ List<String> chainKeys =
+ ChainTableUtils.chainPartitionKeys(options, tableSchema.partitionKeys());
+ int chainFieldCount = chainKeys.size();
+ // full partition comparator
+ this.partitionProjector =
+ new ChainPartitionProjector(
+ tableSchema.logicalPartitionType(), chainFieldCount);
+ // chain dimension comparator (compares only the last chainFieldCount fields)
+ this.chainPartitionComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionProjector.chainPartitionType().getFieldTypes());
}
@Override
public ChainTableBatchScan withFilter(Predicate predicate) {
super.withFilter(predicate);
- if (predicate != null) {
+ if (predicate == null) {
+ dataPredicate = null;
+ } else {
Pair<Optional<PartitionPredicate>, List<Predicate>> pair =
PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
predicate,
@@ -217,11 +225,25 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable {
/**
* Builds a plan for chain tables.
*
- * <p>Partitions that exist in the snapshot branch (based on partition predicates only) are
- * treated as complete and are read directly from snapshot, subject to row-level predicates.
- * Partitions that exist only in the delta branch are planned as chain splits by pairing
- * each delta partition with the latest snapshot partition at or before it (if any), so the
- * reader sees a full partition view.
+ * <p>Partitions that exist in the snapshot branch (matched via the user's partition
+ * predicate) are treated as complete and are read directly from snapshot, subject to
+ * row-level predicates.
+ *
+ * <p>Partitions that exist only in the delta branch are planned group-by-group:
+ *
+ * <ol>
+ * <li>Delta partitions are grouped by their <em>group</em> key prefix (the non-chain
+ * partition fields). When {@code chain-table.chain-partition-keys} is not configured,
+ * all partitions fall into a single implicit group.
+ * <li>For each group the nearest earlier snapshot partition is used as an <em>anchor</em>
+ * for the chain merge. The anchor search uses a targeted predicate (group fields
+ * exact-match AND chain < maxChainInGroup) that intentionally bypasses the user's
+ * partition predicate, so the correct anchor is found even when the user queries a
+ * future or non-existent partition.
+ * <li>A {@link ChainSplit} is built that merges the anchor snapshot data with the delta
+ * data in the range {@code (anchorChain, queryChain]}, giving the reader a complete
+ * view of the queried partition.
+ * </ol>
*/
@Override
public Plan plan() {
@@ -258,110 +280,163 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable {
.collect(Collectors.toList());
if (!deltaPartitions.isEmpty()) {
- BinaryRow maxPartition = deltaPartitions.get(deltaPartitions.size() - 1);
- Predicate snapshotPredicate =
- ChainTableUtils.createTriangularPredicate(
- maxPartition,
- partitionConverter,
- builder::equal,
- builder::lessThan);
- PartitionPredicate snapshotPartitionPredicate =
- PartitionPredicate.fromPredicate(
- tableSchema.logicalPartitionType(), snapshotPredicate);
- DataTableScan snapshotPartitionsScan =
- newChainPartitionListingScan(true, snapshotPartitionPredicate);
- List<BinaryRow> candidateSnapshotPartitions =
- snapshotPartitionsScan.listPartitions();
- candidateSnapshotPartitions =
- candidateSnapshotPartitions.stream()
- .sorted(partitionComparator)
- .collect(Collectors.toList());
- Map<BinaryRow, BinaryRow> partitionMapping =
- ChainTableUtils.findFirstLatestPartitions(
- deltaPartitions, candidateSnapshotPartitions, partitionComparator);
- for (Map.Entry<BinaryRow, BinaryRow> partitionParis : partitionMapping.entrySet()) {
- DataTableScan snapshotScan = newFilteredScan(true);
- DataTableScan deltaScan = newFilteredScan(false);
- if (partitionParis.getValue() == null) {
- List<Predicate> predicates = new ArrayList<>();
- predicates.add(
- ChainTableUtils.createTriangularPredicate(
- partitionParis.getKey(),
- partitionConverter,
- builder::equal,
- builder::lessThan));
- predicates.add(
- ChainTableUtils.createLinearPredicate(
- partitionParis.getKey(),
- partitionConverter,
- builder::equal));
- deltaScan.withPartitionFilter(PredicateBuilder.or(predicates));
- } else {
- List<BinaryRow> selectedDeltaPartitions =
- ChainTableUtils.getDeltaPartitions(
- partitionParis.getValue(),
- partitionParis.getKey(),
- tableSchema.partitionKeys(),
- tableSchema.logicalPartitionType(),
- options,
- partitionComparator,
- partitionComputer);
- deltaScan.withPartitionFilter(selectedDeltaPartitions);
- }
- List<Split> subSplits = deltaScan.plan().splits();
- Set<String> snapshotFileNames = new HashSet<>();
- if (partitionParis.getValue() != null) {
- snapshotScan.withPartitionFilter(
- Collections.singletonList(partitionParis.getValue()));
- List<Split> mainSubSplits = snapshotScan.plan().splits();
- snapshotFileNames =
- mainSubSplits.stream()
- .flatMap(
- s ->
- ((DataSplit) s)
- .dataFiles().stream()
- .map(
- DataFileMeta
- ::fileName))
- .collect(Collectors.toSet());
- subSplits.addAll(mainSubSplits);
- }
- Map<Integer, List<DataSplit>> bucketSplits = new LinkedHashMap<>();
- for (Split split : subSplits) {
- DataSplit dataSplit = (DataSplit) split;
- Integer totalBuckets = dataSplit.totalBuckets();
- checkNotNull(totalBuckets);
- checkArgument(
- totalBuckets == options.bucket(),
- "Inconsistent bucket num " + dataSplit.bucket());
- bucketSplits
- .computeIfAbsent(dataSplit.bucket(), k -> new ArrayList<>())
- .add(dataSplit);
- }
- for (Map.Entry<Integer, List<DataSplit>> entry : bucketSplits.entrySet()) {
- HashMap<String, String> fileBucketPathMapping = new HashMap<>();
- HashMap<String, String> fileBranchMapping = new HashMap<>();
- List<DataSplit> splitList = entry.getValue();
- for (DataSplit dataSplit : splitList) {
- for (DataFileMeta file : dataSplit.dataFiles()) {
- fileBucketPathMapping.put(file.fileName(), dataSplit.bucketPath());
- String branch =
- snapshotFileNames.contains(file.fileName())
- ? options.scanFallbackSnapshotBranch()
- : options.scanFallbackDeltaBranch();
- fileBranchMapping.put(file.fileName(), branch);
+ // Group delta partitions by group key.
+ // When groupFieldCount == 0, extractGroupPartition returns a zero-field BinaryRow,
+ // so all partitions are placed in a single group.
+ Map<BinaryRow, List<BinaryRow>> groupedDeltaPartitions = new LinkedHashMap<>();
+ for (BinaryRow partition : deltaPartitions) {
+ BinaryRow groupKey = partitionProjector.extractGroupPartition(partition);
+ groupedDeltaPartitions
+ .computeIfAbsent(groupKey, k -> new ArrayList<>())
+ .add(partition);
+ }
+
+ for (List<BinaryRow> deltaPartitionsInGroup : groupedDeltaPartitions.values()) {
+
+ // Sort delta by chain dimension ascending.
+ // chainPartitionForCompare avoids copying BinaryRow in the comparator hot path.
+ deltaPartitionsInGroup.sort(
+ (a, b) ->
+ chainPartitionComparator.compare(
+ partitionProjector.chainPartitionForCompare(a),
+ partitionProjector.chainPartitionForCompare(b)));
+
+ // Build a targeted snapshot-anchor predicate:
+ // group fields exact-match AND chain < maxChainInGroup
+ // When groupFieldCount == 0 this degenerates to the plain triangular predicate
+ // (equivalent to the former planWithoutGroupPartition snapshot listing).
+ BinaryRow maxDeltaInGroup =
+ deltaPartitionsInGroup.get(deltaPartitionsInGroup.size() - 1);
+ Predicate snapshotSearchPred =
+ ChainTableUtils.createGroupChainPredicate(
+ maxDeltaInGroup,
+ partitionConverter,
+ partitionProjector.groupFieldCount(),
+ builder::equal,
+ builder::lessThan);
+ PartitionPredicate snapshotAnchorPredicate =
+ PartitionPredicate.fromPredicate(
+ tableSchema.logicalPartitionType(), snapshotSearchPred);
+
+ // List snapshot partitions for this group, sorted by chain dimension.
+ List<BinaryRow> snapshotPartitionsInGroup =
+ newChainPartitionListingScan(true, snapshotAnchorPredicate)
+ .listPartitions().stream()
+ .sorted(
+ (a, b) ->
+ chainPartitionComparator.compare(
+ partitionProjector
+ .chainPartitionForCompare(a),
+ partitionProjector
+ .chainPartitionForCompare(b)))
+ .collect(Collectors.toList());
+
+ // Find delta → snapshot mapping (for each delta partition, find the nearest
+ // earlier
+ // snapshot)
+ // Note: uses chainPartitionComparator, comparing only the chain dimension
+ Map<BinaryRow, BinaryRow> partitionMapping =
+ ChainTableUtils.findFirstLatestPartitionsWithProjector(
+ deltaPartitionsInGroup,
+ snapshotPartitionsInGroup,
+ chainPartitionComparator,
+ partitionProjector);
+
+ for (Map.Entry<BinaryRow, BinaryRow> partitionPairs :
+ partitionMapping.entrySet()) {
+ BinaryRow deltaPartition = partitionPairs.getKey();
+ BinaryRow snapshotPartition = partitionPairs.getValue();
+
+ DataTableScan snapshotScan = newFilteredScan(true);
+ DataTableScan deltaScan = newFilteredScan(false);
+
+ if (snapshotPartition == null) {
+ List<Predicate> predicates = new ArrayList<>();
+ predicates.add(
+ ChainTableUtils.createGroupChainPredicate(
+ deltaPartition,
+ partitionConverter,
+ partitionProjector.groupFieldCount(),
+ builder::equal,
+ builder::lessThan));
+ predicates.add(
+ ChainTableUtils.createLinearPredicate(
+ deltaPartition, partitionConverter, builder::equal));
+ deltaScan.withPartitionFilter(PredicateBuilder.or(predicates));
+ } else {
+ List<BinaryRow> selectedDeltaPartitions =
+ ChainTableUtils.getDeltaPartitionsWithProjector(
+ snapshotPartition,
+ deltaPartition,
+ options,
+ chainPartitionComparator,
+ partitionProjector);
+ deltaScan.withPartitionFilter(selectedDeltaPartitions);
+ }
+
+ List<Split> subSplits = deltaScan.plan().splits();
+ Set<String> snapshotFileNames = new HashSet<>();
+ if (partitionPairs.getValue() != null) {
+ snapshotScan.withPartitionFilter(
+ Collections.singletonList(partitionPairs.getValue()));
+ List<Split> mainSubSplits = snapshotScan.plan().splits();
+ snapshotFileNames =
+ mainSubSplits.stream()
+ .flatMap(
+ s ->
+ ((DataSplit) s)
+ .dataFiles().stream()
+ .map(
+ DataFileMeta
+ ::fileName))
+ .collect(Collectors.toSet());
+ subSplits.addAll(mainSubSplits);
+ }
+ Map<Integer, List<DataSplit>> bucketSplits = new LinkedHashMap<>();
+ Integer bucketInAll = null;
+ for (Split split : subSplits) {
+ DataSplit dataSplit = (DataSplit) split;
+ Integer totalBuckets = dataSplit.totalBuckets();
+ checkNotNull(totalBuckets);
+ if (bucketInAll == null) {
+ bucketInAll = totalBuckets;
+ } else {
+ checkArgument(
+ totalBuckets == bucketInAll,
+ "Inconsistent bucket num " + dataSplit.bucket());
+ }
+
+ bucketSplits
+ .computeIfAbsent(dataSplit.bucket(), k -> new ArrayList<>())
+ .add(dataSplit);
+ }
+ for (Map.Entry<Integer, List<DataSplit>> entry : bucketSplits.entrySet()) {
+ HashMap<String, String> fileBucketPathMapping = new HashMap<>();
+ HashMap<String, String> fileBranchMapping = new HashMap<>();
+ List<DataSplit> splitList = entry.getValue();
+ for (DataSplit dataSplit : splitList) {
+ for (DataFileMeta file : dataSplit.dataFiles()) {
+ fileBucketPathMapping.put(
+ file.fileName(), dataSplit.bucketPath());
+ String branch =
+ snapshotFileNames.contains(file.fileName())
+ ? options.scanFallbackSnapshotBranch()
+ : options.scanFallbackDeltaBranch();
+ fileBranchMapping.put(file.fileName(), branch);
+ }
}
+ ChainSplit split =
+ new ChainSplit(
+ partitionPairs.getKey(),
+ entry.getValue().stream()
+ .flatMap(
+ dataSplit ->
+ dataSplit.dataFiles().stream())
+ .collect(Collectors.toList()),
+ fileBranchMapping,
+ fileBucketPathMapping);
+ splits.add(split);
}
- ChainSplit split =
- new ChainSplit(
- partitionParis.getKey(),
- entry.getValue().stream()
- .flatMap(
- datsSplit -> datsSplit.dataFiles().stream())
- .collect(Collectors.toList()),
- fileBranchMapping,
- fileBucketPathMapping);
- splits.add(split);
}
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ChainPartitionProjector.java b/paimon-core/src/main/java/org/apache/paimon/utils/ChainPartitionProjector.java
new file mode 100644
index 0000000000..58fd7eb684
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ChainPartitionProjector.java
@@ -0,0 +1,160 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.Projection;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.types.RowType;
+
+import java.util.stream.IntStream;
+
+/**
+ * Utility to project full partition rows into group (non-chain) and chain parts, and to combine
+ * them back into full partition rows.
+ *
+ * <p>Given partition keys [region, date] with chain partition keys = [date]:
+ *
+ * <ul>
+ * <li>groupPartition = projection of [region]
+ * <li>chainPartition = projection of [date]
+ * </ul>
+ */
+public class ChainPartitionProjector {
+ private final Projection groupProjection;
+ private final Projection chainProjection;
+ private final InternalRowSerializer fullSerializer;
+ private final RowType fullPartitionType;
+ private final int groupFieldCount;
+ private final int chainFieldCount;
+
+ // Cached projected RowTypes to avoid repeated IntStream allocation on each call.
+ private final RowType groupType;
+ private final RowType chainType;
+
+ // Pre-built FieldGetter arrays for combinePartition
+ private final InternalRow.FieldGetter[] groupFieldGetters;
+ private final InternalRow.FieldGetter[] chainFieldGetters;
+
+ /**
+ * @param fullPartitionType the full partition RowType
+ * @param chainFieldCount number of chain partition fields (counted from the end)
+ */
+ public ChainPartitionProjector(RowType fullPartitionType, int chainFieldCount) {
+ this.fullPartitionType = fullPartitionType;
+ int totalFields = fullPartitionType.getFieldCount();
+ this.groupFieldCount = totalFields - chainFieldCount;
+ this.chainFieldCount = chainFieldCount;
+
+ // group: first groupFieldCount fields
+ int[] groupIndices = new int[groupFieldCount];
+ for (int i = 0; i < groupFieldCount; i++) {
+ groupIndices[i] = i;
+ }
+ this.groupProjection = CodeGenUtils.newProjection(fullPartitionType, groupIndices);
+
+ // chain: last chainFieldCount fields
+ int[] chainIndices = new int[chainFieldCount];
+ for (int i = 0; i < chainFieldCount; i++) {
+ chainIndices[i] = groupFieldCount + i;
+ }
+ this.chainProjection = CodeGenUtils.newProjection(fullPartitionType, chainIndices);
+
+ this.fullSerializer = new InternalRowSerializer(fullPartitionType);
+
+ // Cache projected RowTypes once; used by groupPartitionType()/chainPartitionType()
+ // and the FieldGetter initialization below.
+ this.groupType = fullPartitionType.project(IntStream.range(0, groupFieldCount).toArray());
+ this.chainType =
+ fullPartitionType.project(
+ IntStream.range(groupFieldCount, groupFieldCount + chainFieldCount)
+ .toArray());
+
+ // Pre-build FieldGetters for type-safe field access from group/chain BinaryRow
+ this.groupFieldGetters = new InternalRow.FieldGetter[groupFieldCount];
+ for (int i = 0; i < groupFieldCount; i++) {
+ groupFieldGetters[i] = InternalRow.createFieldGetter(groupType.getTypeAt(i), i);
+ }
+
+ this.chainFieldGetters = new InternalRow.FieldGetter[chainFieldCount];
+ for (int i = 0; i < chainFieldCount; i++) {
+ chainFieldGetters[i] = InternalRow.createFieldGetter(chainType.getTypeAt(i), i);
+ }
+ }
+
+ /** Extract group partitions from full partition. */
+ public BinaryRow extractGroupPartition(BinaryRow fullPartition) {
+ return groupProjection.apply(fullPartition).copy();
+ }
+
+ /** Extract chain partitions from full partition. */
+ public BinaryRow extractChainPartition(BinaryRow fullPartition) {
+ return chainProjection.apply(fullPartition).copy();
+ }
+
+ /**
+ * Combines the group part and chain part back into a full partition row. Uses
+ * InternalRow.createFieldGetter for type-safe field value extraction.
+ */
+ public BinaryRow combinePartition(BinaryRow groupPart, BinaryRow chainPart) {
+ GenericRow combined = new GenericRow(fullPartitionType.getFieldCount());
+ for (int i = 0; i < groupFieldCount; i++) {
+ combined.setField(i, groupFieldGetters[i].getFieldOrNull(groupPart));
+ }
+ for (int i = 0; i < chainFieldCount; i++) {
+ combined.setField(groupFieldCount + i, chainFieldGetters[i].getFieldOrNull(chainPart));
+ }
+ return fullSerializer.toBinaryRow(combined).copy();
+ }
+
+ public boolean hasGroupPartition() {
+ return groupFieldCount > 0;
+ }
+
+ public int groupFieldCount() {
+ return groupFieldCount;
+ }
+
+ public int chainFieldCount() {
+ return chainFieldCount;
+ }
+
+ public RowType groupPartitionType() {
+ return groupType;
+ }
+
+ public RowType chainPartitionType() {
+ return chainType;
+ }
+
+ /**
+ * Returns a projection of the chain fields from the given full-partition row <em>without
+ * copying</em> the underlying storage.
+ *
+ * <p>The returned {@link InternalRow} shares the backing bytes of {@code fullPartition} and
+ * must not be retained across mutations of that row. Use this method only for transient,
+ * read-only comparisons to avoid unnecessary {@link BinaryRow} allocations.
+ */
+ public InternalRow chainPartitionForCompare(BinaryRow fullPartition) {
+ return chainProjection.apply(fullPartition);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
index f6c573440f..6ccbbbc8fc 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/ChainTableUtils.java
@@ -225,4 +225,136 @@ public class ChainTableUtils {
}
return table;
}
+
+ public static List<String> chainPartitionKeys(
+ CoreOptions options, List<String> allPartitionKeys) {
+ List<String> chainPartitionKeys = options.chainTableChainPartitionKeys();
+ if (chainPartitionKeys == null) {
+ return allPartitionKeys;
+ }
+ return chainPartitionKeys;
+ }
+
+ /**
+ * Within the same group, find the nearest smaller partition in the target list for each source
+ * partition, comparing only on the chain dimension.
+ *
+ * @param sortedSourcePartitions full partitions (sorted by chain dimension)
+ * @param sortedTargetPartitions full partitions (sorted by chain dimension, same group)
+ * @param chainComparator compares chain dimension only
+ * @param projector partition projector
+ * @return source → target mapping, target may be null
+ */
+ public static Map<BinaryRow, BinaryRow> findFirstLatestPartitionsWithProjector(
+ List<BinaryRow> sortedSourcePartitions,
+ List<BinaryRow> sortedTargetPartitions,
+ RecordComparator chainComparator,
+ ChainPartitionProjector projector) {
+
+ Map<BinaryRow, BinaryRow> partitionMapping = new HashMap<>();
+ int targetIndex = 0;
+
+ for (BinaryRow sourceRow : sortedSourcePartitions) {
+ BinaryRow sourceChain = projector.extractChainPartition(sourceRow);
+ while (targetIndex < sortedTargetPartitions.size()) {
+ BinaryRow targetChain =
+ projector.extractChainPartition(sortedTargetPartitions.get(targetIndex));
+ if (chainComparator.compare(targetChain, sourceChain) < 0) {
+ targetIndex++;
+ } else {
+ break;
+ }
+ }
+ BinaryRow firstSmaller =
+ (targetIndex > 0) ? sortedTargetPartitions.get(targetIndex - 1) : null;
+ partitionMapping.put(sourceRow, firstSmaller);
+ }
+ return partitionMapping;
+ }
+
+ /**
+ * Generates the list of delta partitions in the range (beginPartition, endPartition].
+ * Enumerates time range only on the chain dimension; the group dimension stays unchanged.
+ */
+ public static List<BinaryRow> getDeltaPartitionsWithProjector(
+ BinaryRow beginPartition,
+ BinaryRow endPartition,
+ CoreOptions options,
+ RecordComparator chainPartitionComparator,
+ ChainPartitionProjector projector) {
+
+ // Extract the chain parts from begin/end
+ BinaryRow beginChain = projector.extractChainPartition(beginPartition);
+ BinaryRow endChain = projector.extractChainPartition(endPartition);
+
+ // Build chain-dimension RowType, column names, PartitionComputer
+ RowType chainPartType = projector.chainPartitionType();
+ List<String> chainPartitionColumns = chainPartType.getFieldNames();
+ InternalRowPartitionComputer chainPartitionComputer =
+ new InternalRowPartitionComputer(
+ options.partitionDefaultName(),
+ chainPartType,
+ chainPartitionColumns.toArray(new String[0]),
+ options.legacyPartitionName());
+
+ // Reuse existing getDeltaPartitions to enumerate on chain dimension
+ List<BinaryRow> chainOnlyDeltas =
+ getDeltaPartitions(
+ beginChain,
+ endChain,
+ chainPartitionColumns,
+ chainPartType,
+ options,
+ chainPartitionComparator,
+ chainPartitionComputer);
+
+ // Combine each chain-only BinaryRow with the group part into a full partition
+ BinaryRow groupPart = projector.extractGroupPartition(beginPartition);
+ List<BinaryRow> fullDeltas = new ArrayList<>(chainOnlyDeltas.size());
+ for (BinaryRow chainDelta : chainOnlyDeltas) {
+ fullDeltas.add(projector.combinePartition(groupPart, chainDelta));
+ }
+ return fullDeltas;
+ }
+
+ /**
+ * Builds a compound predicate: group fields exact match AND chain fields triangular range.
+ *
+ * @param fullPartition full partition row
+ * @param converter converter for the full partition
+ * @param groupFieldCount number of group fields
+ * @param innerFunc equality function (field_index, value) → Predicate
+ * @param outerFunc range function (field_index, value) → Predicate
+ */
+ public static Predicate createGroupChainPredicate(
+ BinaryRow fullPartition,
+ RowDataToObjectArrayConverter converter,
+ int groupFieldCount,
+ BiFunction<Integer, Object, Predicate> innerFunc,
+ BiFunction<Integer, Object, Predicate> outerFunc) {
+
+ Object[] allValues = converter.convert(fullPartition);
+ List<Predicate> conditions = new ArrayList<>();
+
+ for (int i = 0; i < groupFieldCount; i++) {
+ conditions.add(innerFunc.apply(i, allValues[i]));
+ }
+
+ List<Predicate> chainFieldPredicates = new ArrayList<>();
+ int totalFields = converter.getArity();
+ for (int i = groupFieldCount; i < totalFields; i++) {
+ List<Predicate> andConditions = new ArrayList<>();
+ for (int j = groupFieldCount; j < i; j++) {
+ andConditions.add(innerFunc.apply(j, allValues[j]));
+ }
+ andConditions.add(outerFunc.apply(i, allValues[i]));
+ chainFieldPredicates.add(PredicateBuilder.and(andConditions));
+ }
+
+ if (!chainFieldPredicates.isEmpty()) {
+ conditions.add(PredicateBuilder.or(chainFieldPredicates));
+ }
+
+ return PredicateBuilder.and(conditions);
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/ChainPartitionProjectorTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/ChainPartitionProjectorTest.java
new file mode 100644
index 0000000000..8f7cd36393
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/ChainPartitionProjectorTest.java
@@ -0,0 +1,432 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.utils;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test class for {@link ChainPartitionProjector}. */
+public class ChainPartitionProjectorTest {
+
+ // ========================== Helper Methods ==========================
+
+ /** Build a BinaryRow with given string values for a STRING-only RowType. */
+ private static BinaryRow buildStringRow(String... values) {
+ BinaryRow row = new BinaryRow(values.length);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ for (int i = 0; i < values.length; i++) {
+ if (values[i] == null) {
+ writer.setNullAt(i);
+ } else {
+ writer.writeString(i, BinaryString.fromString(values[i]));
+ }
+ }
+ writer.complete();
+ return row;
+ }
+
+ /** Build a BinaryRow with mixed types: string at pos 0, int at pos 1. */
+ private static BinaryRow buildStringIntRow(String strVal, int intVal) {
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeString(0, BinaryString.fromString(strVal));
+ writer.writeInt(1, intVal);
+ writer.complete();
+ return row;
+ }
+
+ /** Extract a string value from a BinaryRow at the given position. */
+ private static String getString(BinaryRow row, int pos) {
+ return row.getString(pos).toString();
+ }
+
+ // ========================== Test: Two STRING fields (region, date) ==========================
+
+ @Test
+ public void testExtractGroupAndChainWithTwoStringFields() {
+ // partition keys: (region, date), chain partition keys: (date)
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow fullPartition = buildStringRow("US", "20250726");
+
+ // extract group
+ BinaryRow groupPart = projector.extractGroupPartition(fullPartition);
+ assertThat(groupPart.getFieldCount()).isEqualTo(1);
+ assertThat(getString(groupPart, 0)).isEqualTo("US");
+
+ // extract chain
+ BinaryRow chainPart = projector.extractChainPartition(fullPartition);
+ assertThat(chainPart.getFieldCount()).isEqualTo(1);
+ assertThat(getString(chainPart, 0)).isEqualTo("20250726");
+ }
+
+ @Test
+ public void testCombinePartitionWithTwoStringFields() {
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow groupPart = buildStringRow("EU");
+ BinaryRow chainPart = buildStringRow("20250801");
+
+ BinaryRow combined = projector.combinePartition(groupPart, chainPart);
+ assertThat(combined.getFieldCount()).isEqualTo(2);
+ assertThat(getString(combined, 0)).isEqualTo("EU");
+ assertThat(getString(combined, 1)).isEqualTo("20250801");
+ }
+
+ @Test
+ public void testRoundTripExtractThenCombineWithTwoStringFields() {
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow original = buildStringRow("CN", "20250720");
+
+ BinaryRow groupPart = projector.extractGroupPartition(original);
+ BinaryRow chainPart = projector.extractChainPartition(original);
+ BinaryRow reconstructed = projector.combinePartition(groupPart, chainPart);
+
+ assertThat(getString(reconstructed, 0)).isEqualTo("CN");
+ assertThat(getString(reconstructed, 1)).isEqualTo("20250720");
+
+ // verify binary equality via serialization
+ InternalRowSerializer serializer = new InternalRowSerializer(fullType);
+ assertThat(serializer.toBinaryRow(reconstructed).copy())
+ .isEqualTo(serializer.toBinaryRow(original).copy());
+ }
+
+ // ========================== Test: Three STRING fields ==========================
+
+ @Test
+ public void testThreeStringFieldsWithTwoChain() {
+ // partition keys: (country, city, date), chain partition keys: (city, date)
+ RowType fullType =
+ RowType.builder()
+ .field("country", DataTypes.STRING().notNull())
+ .field("city", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 2);
+
+ assertThat(projector.groupFieldCount()).isEqualTo(1);
+ assertThat(projector.chainFieldCount()).isEqualTo(2);
+ assertThat(projector.hasGroupPartition()).isTrue();
+
+ BinaryRow fullPartition = buildStringRow("China", "Beijing", "20250801");
+
+ // group = (country)
+ BinaryRow groupPart = projector.extractGroupPartition(fullPartition);
+ assertThat(groupPart.getFieldCount()).isEqualTo(1);
+ assertThat(getString(groupPart, 0)).isEqualTo("China");
+
+ // chain = (city, date)
+ BinaryRow chainPart = projector.extractChainPartition(fullPartition);
+ assertThat(chainPart.getFieldCount()).isEqualTo(2);
+ assertThat(getString(chainPart, 0)).isEqualTo("Beijing");
+ assertThat(getString(chainPart, 1)).isEqualTo("20250801");
+
+ // round-trip
+ BinaryRow combined = projector.combinePartition(groupPart, chainPart);
+ assertThat(getString(combined, 0)).isEqualTo("China");
+ assertThat(getString(combined, 1)).isEqualTo("Beijing");
+ assertThat(getString(combined, 2)).isEqualTo("20250801");
+ }
+
+ @Test
+ public void testThreeStringFieldsWithOneChain() {
+ // partition keys: (country, city, date), chain partition keys: (date)
+ RowType fullType =
+ RowType.builder()
+ .field("country", DataTypes.STRING().notNull())
+ .field("city", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ assertThat(projector.groupFieldCount()).isEqualTo(2);
+ assertThat(projector.chainFieldCount()).isEqualTo(1);
+
+ BinaryRow fullPartition = buildStringRow("US", "NYC", "20250901");
+
+ // group = (country, city)
+ BinaryRow groupPart = projector.extractGroupPartition(fullPartition);
+ assertThat(groupPart.getFieldCount()).isEqualTo(2);
+ assertThat(getString(groupPart, 0)).isEqualTo("US");
+ assertThat(getString(groupPart, 1)).isEqualTo("NYC");
+
+ // chain = (date)
+ BinaryRow chainPart = projector.extractChainPartition(fullPartition);
+ assertThat(chainPart.getFieldCount()).isEqualTo(1);
+ assertThat(getString(chainPart, 0)).isEqualTo("20250901");
+ }
+
+ // ========================== Test: All partitions are chain (no group)
+ // ==========================
+
+ @Test
+ public void testAllChainWithNoGroup() {
+ // partition keys: (date), chain partition keys: (date)
+ RowType fullType = RowType.builder().field("date", DataTypes.STRING().notNull()).build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ assertThat(projector.groupFieldCount()).isEqualTo(0);
+ assertThat(projector.chainFieldCount()).isEqualTo(1);
+ assertThat(projector.hasGroupPartition()).isFalse();
+
+ BinaryRow fullPartition = buildStringRow("20250726");
+
+ // group is empty
+ BinaryRow groupPart = projector.extractGroupPartition(fullPartition);
+ assertThat(groupPart.getFieldCount()).isEqualTo(0);
+
+ // chain is the whole partition
+ BinaryRow chainPart = projector.extractChainPartition(fullPartition);
+ assertThat(chainPart.getFieldCount()).isEqualTo(1);
+ assertThat(getString(chainPart, 0)).isEqualTo("20250726");
+ }
+
+ @Test
+ public void testAllChainWithTwoFields() {
+ // partition keys: (dt, hour), chain partition keys: (dt, hour)
+ RowType fullType =
+ RowType.builder()
+ .field("dt", DataTypes.STRING().notNull())
+ .field("hour", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 2);
+
+ assertThat(projector.hasGroupPartition()).isFalse();
+ assertThat(projector.groupFieldCount()).isEqualTo(0);
+ assertThat(projector.chainFieldCount()).isEqualTo(2);
+
+ BinaryRow fullPartition = buildStringRow("20250810", "14");
+
+ BinaryRow chainPart = projector.extractChainPartition(fullPartition);
+ assertThat(getString(chainPart, 0)).isEqualTo("20250810");
+ assertThat(getString(chainPart, 1)).isEqualTo("14");
+ }
+
+ // ========================== Test: Mixed types (STRING + INT) ==========================
+
+ @Test
+ public void testMixedTypesWithStringAndInt() {
+ // partition keys: (region STRING, bucket_id INT), chain partition keys: (bucket_id)
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("bucket_id", DataTypes.INT().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow fullPartition = buildStringIntRow("US", 42);
+
+ // group = (region)
+ BinaryRow groupPart = projector.extractGroupPartition(fullPartition);
+ assertThat(groupPart.getFieldCount()).isEqualTo(1);
+ assertThat(getString(groupPart, 0)).isEqualTo("US");
+
+ // chain = (bucket_id)
+ BinaryRow chainPart = projector.extractChainPartition(fullPartition);
+ assertThat(chainPart.getFieldCount()).isEqualTo(1);
+ assertThat(chainPart.getInt(0)).isEqualTo(42);
+
+ // combine
+ BinaryRow combined = projector.combinePartition(groupPart, chainPart);
+ assertThat(getString(combined, 0)).isEqualTo("US");
+ assertThat(combined.getInt(1)).isEqualTo(42);
+ }
+
+ // ========================== Test: RowType accessors ==========================
+
+ @Test
+ public void testGroupPartitionType() {
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("city", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ RowType groupType = projector.groupPartitionType();
+ assertThat(groupType.getFieldCount()).isEqualTo(2);
+ assertThat(groupType.getFieldNames().get(0)).isEqualTo("region");
+ assertThat(groupType.getFieldNames().get(1)).isEqualTo("city");
+
+ RowType chainType = projector.chainPartitionType();
+ assertThat(chainType.getFieldCount()).isEqualTo(1);
+ assertThat(chainType.getFieldNames().get(0)).isEqualTo("date");
+ }
+
+ @Test
+ public void testGroupPartitionTypeWithNoGroup() {
+ RowType fullType = RowType.builder().field("date", DataTypes.STRING().notNull()).build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ RowType groupType = projector.groupPartitionType();
+ assertThat(groupType.getFieldCount()).isEqualTo(0);
+
+ RowType chainType = projector.chainPartitionType();
+ assertThat(chainType.getFieldCount()).isEqualTo(1);
+ assertThat(chainType.getFieldNames().get(0)).isEqualTo("date");
+ }
+
+ // ========================== Test: Multiple partitions grouping ==========================
+
+ @Test
+ public void testGroupPartitionConsistency() {
+ // Verify that partitions with same group prefix produce equal group projections
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow partition1 = buildStringRow("US", "20250721");
+ BinaryRow partition2 = buildStringRow("US", "20250722");
+ BinaryRow partition3 = buildStringRow("EU", "20250721");
+
+ BinaryRow group1 = projector.extractGroupPartition(partition1).copy();
+ BinaryRow group2 = projector.extractGroupPartition(partition2).copy();
+ BinaryRow group3 = projector.extractGroupPartition(partition3).copy();
+
+ // Same region => equal group partitions
+ assertThat(group1).isEqualTo(group2);
+
+ // Different region => different group partitions
+ assertThat(group1).isNotEqualTo(group3);
+ }
+
+ @Test
+ public void testChainPartitionOrdering() {
+ // Verify that chain projections can be compared correctly
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow partition1 = buildStringRow("US", "20250720");
+ BinaryRow partition2 = buildStringRow("US", "20250726");
+ BinaryRow partition3 = buildStringRow("EU", "20250720");
+
+ BinaryRow chain1 = projector.extractChainPartition(partition1).copy();
+ BinaryRow chain2 = projector.extractChainPartition(partition2).copy();
+ BinaryRow chain3 = projector.extractChainPartition(partition3).copy();
+
+ // Same date, different region => equal chain partitions
+ assertThat(chain1).isEqualTo(chain3);
+
+ // Different date => different chain partitions
+ assertThat(chain1).isNotEqualTo(chain2);
+
+ // Verify string ordering: "20250720" < "20250726"
+ assertThat(chain1.getString(0).compareTo(chain2.getString(0))).isLessThan(0);
+ }
+
+ // ========================== Test: Nullable fields ==========================
+
+ @Test
+ public void testNullableGroupField() {
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ // build a row with null region
+ BinaryRow row = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.setNullAt(0);
+ writer.writeString(1, BinaryString.fromString("20250801"));
+ writer.complete();
+
+ BinaryRow groupPart = projector.extractGroupPartition(row);
+ assertThat(groupPart.isNullAt(0)).isTrue();
+
+ BinaryRow chainPart = projector.extractChainPartition(row);
+ assertThat(getString(chainPart, 0)).isEqualTo("20250801");
+
+ // combine should preserve null
+ BinaryRow combined = projector.combinePartition(groupPart, chainPart);
+ assertThat(combined.isNullAt(0)).isTrue();
+ assertThat(getString(combined, 1)).isEqualTo("20250801");
+ }
+
+ // ========================== Test: Combine with different group/chain sources
+
+ @Test
+ public void testCombineFromDifferentSources() {
+ // Simulate: take group from one partition, chain from another
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+
+ BinaryRow partitionA = buildStringRow("US", "20250720");
+ BinaryRow partitionB = buildStringRow("EU", "20250801");
+
+ // Take group from A (US), chain from B (20250801)
+ BinaryRow groupFromA = projector.extractGroupPartition(partitionA);
+ BinaryRow chainFromB = projector.extractChainPartition(partitionB);
+
+ BinaryRow mixed = projector.combinePartition(groupFromA, chainFromB);
+ assertThat(getString(mixed, 0)).isEqualTo("US");
+ assertThat(getString(mixed, 1)).isEqualTo("20250801");
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
index 99ba22c451..051323e180 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/ChainTableUtilsTest.java
@@ -18,16 +18,19 @@
package org.apache.paimon.utils;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.assertj.core.util.Lists;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -40,6 +43,7 @@ import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
/** Test class for {@link org.apache.paimon.utils.ChainTableUtils}. */
@@ -56,6 +60,21 @@ public class ChainTableUtilsTest {
return builder.toString();
}
+ /** Extract a string value from a BinaryRow at the given position. */
+ private static String getString(BinaryRow row, int pos) {
+ return row.getString(pos).toString();
+ }
+
+ private static BinaryRow row(List<String> values) {
+ BinaryRow row = new BinaryRow(values.size());
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ for (int i = 0; i < values.size(); i++) {
+ writer.writeString(i, BinaryString.fromString(values.get(i)));
+ }
+ writer.complete();
+ return row;
+ }
+
@Test
public void testFindFirstLatestPartitions() {
String[] snapshotPartitions = {"20251105", "20251101"};
@@ -189,4 +208,517 @@ public class ChainTableUtilsTest {
},
partitionValues);
}
+
+ // ========================== Tests for findFirstLatestPartitionsWithProjector
+ // ==========================
+
+ @Test
+ public void testFindFirstLatestPartitionsWithProjectorWithBasicGrouping() {
+ // partition keys: (region, date), chain keys: (date)
+ // Snapshot: (US,20250720), (US,20250725)
+ // Delta: (US,20250722), (US,20250727)
+ // Expected: 20250722 → 20250720, 20250727 → 20250725
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ // Snapshot partitions, pre-sorted by chain field (date)
+ BinaryRow snap1 = row(Lists.newArrayList("US", "20250720"));
+ BinaryRow snap2 = row(Lists.newArrayList("US", "20250725"));
+ List<BinaryRow> snapshotPartitions = Arrays.asList(snap1, snap2);
+
+ // Delta partitions, pre-sorted by chain field (date)
+ BinaryRow delta1 = row(Lists.newArrayList("US", "20250722"));
+ BinaryRow delta2 = row(Lists.newArrayList("US", "20250727"));
+ List<BinaryRow> deltaPartitions = Arrays.asList(delta1, delta2);
+
+ Map<BinaryRow, BinaryRow> mapping =
+ ChainTableUtils.findFirstLatestPartitionsWithProjector(
+ deltaPartitions, snapshotPartitions, chainComparator, projector);
+
+ // delta (US,20250722) → snapshot (US,20250720)
+ BinaryRow matched1 = mapping.get(delta1);
+ assertThat(matched1).isNotNull();
+ assertThat(getString(matched1, 0)).isEqualTo("US");
+ assertThat(getString(matched1, 1)).isEqualTo("20250720");
+
+ // delta (US,20250727) → snapshot (US,20250725)
+ BinaryRow matched2 = mapping.get(delta2);
+ assertThat(matched2).isNotNull();
+ assertThat(getString(matched2, 0)).isEqualTo("US");
+ assertThat(getString(matched2, 1)).isEqualTo("20250725");
+ }
+
+ @Test
+ public void testFindFirstLatestPartitionsWithProjectorWithNoMatchingSnapshot() {
+ // Delta partition is earlier than all snapshots → null mapping
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ List<BinaryRow> snapshotPartitions =
+ Arrays.asList(row(Lists.newArrayList("US", "20250725")));
+ BinaryRow delta1 = row(Lists.newArrayList("US", "20250720"));
+ List<BinaryRow> deltaPartitions = Arrays.asList(delta1);
+
+ Map<BinaryRow, BinaryRow> mapping =
+ ChainTableUtils.findFirstLatestPartitionsWithProjector(
+ deltaPartitions, snapshotPartitions, chainComparator, projector);
+
+ // delta 20250720 < snapshot 20250725, no earlier snapshot → null
+ assertThat(mapping.get(delta1)).isNull();
+ }
+
+ @Test
+ public void testFindFirstLatestPartitionsWithProjectorWithEmptySnapshot() {
+ // No snapshot partitions at all → all mappings are null
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ BinaryRow delta1 = row(Lists.newArrayList("US", "20250720"));
+ BinaryRow delta2 = row(Lists.newArrayList("US", "20250722"));
+ List<BinaryRow> deltaPartitions = Arrays.asList(delta1, delta2);
+
+ Map<BinaryRow, BinaryRow> mapping =
+ ChainTableUtils.findFirstLatestPartitionsWithProjector(
+ deltaPartitions, new ArrayList<>(), chainComparator, projector);
+
+ for (BinaryRow delta : deltaPartitions) {
+ assertThat(mapping.get(delta)).isNull();
+ }
+ }
+
+ @Test
+ public void testFindFirstLatestPartitionsWithProjectorWithMultipleChainFields() {
+ // partition keys: (region, dt, hour), chain keys: (dt, hour)
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("dt", DataTypes.STRING().notNull())
+ .field("hour", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 2);
+
+ // Compare chain partition (dt, hour) lexicographically
+ RecordComparator chainComparator =
+ (a, b) -> {
+ int cmp = a.getString(0).compareTo(b.getString(0));
+ if (cmp != 0) {
+ return cmp;
+ }
+ return a.getString(1).compareTo(b.getString(1));
+ };
+
+ // snapshot: (US, 20250720, 12)
+ BinaryRow snap1 = row(Lists.newArrayList("US", "20250720", "12"));
+ List<BinaryRow> snapshotPartitions = Arrays.asList(snap1);
+
+ // delta: (US, 20250720, 10), (US, 20250720, 14) — pre-sorted by chain (dt, hour)
+ BinaryRow delta1 = row(Lists.newArrayList("US", "20250720", "10"));
+ BinaryRow delta2 = row(Lists.newArrayList("US", "20250720", "14"));
+ List<BinaryRow> deltaPartitions = Arrays.asList(delta1, delta2);
+
+ Map<BinaryRow, BinaryRow> mapping =
+ ChainTableUtils.findFirstLatestPartitionsWithProjector(
+ deltaPartitions, snapshotPartitions, chainComparator, projector);
+
+ // delta (US,20250720,10): chain (20250720,10) < snapshot chain (20250720,12) → null
+ assertThat(mapping.get(delta1)).isNull();
+
+ // delta (US,20250720,14): chain (20250720,14) > snapshot chain (20250720,12) → match
+ BinaryRow matched = mapping.get(delta2);
+ assertThat(matched).isNotNull();
+ assertThat(getString(matched, 1)).isEqualTo("20250720");
+ assertThat(getString(matched, 2)).isEqualTo("12");
+ }
+
+ // ========================== Tests for getDeltaPartitionsWithProjector
+ // ==========================
+
+ @Test
+ public void testGetDeltaPartitionsWithProjectorWithDailyPartition() {
+ // partition keys: (region, date), chain keys: (date)
+ // begin: (US, 20250720), end: (US, 20250724)
+ // Expected delta: (US,20250721), (US,20250722), (US,20250723), (US,20250724)
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ Options opts = new Options();
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_PATTERN, "$date");
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd");
+ CoreOptions options = new CoreOptions(opts);
+
+ BinaryRow begin = row(Lists.newArrayList("US", "20250720"));
+ BinaryRow end = row(Lists.newArrayList("US", "20250724"));
+
+ List<BinaryRow> deltas =
+ ChainTableUtils.getDeltaPartitionsWithProjector(
+ begin, end, options, chainComparator, projector);
+
+ // (20250720, 20250724]: 4 days
+ assertThat(deltas).hasSize(4);
+ assertThat(getString(deltas.get(0), 0)).isEqualTo("US");
+ assertThat(getString(deltas.get(0), 1)).isEqualTo("20250721");
+ assertThat(getString(deltas.get(1), 1)).isEqualTo("20250722");
+ assertThat(getString(deltas.get(2), 1)).isEqualTo("20250723");
+ assertThat(getString(deltas.get(3), 1)).isEqualTo("20250724");
+
+ // all should have group = US
+ for (BinaryRow delta : deltas) {
+ assertThat(getString(delta, 0)).isEqualTo("US");
+ }
+ }
+
+ @Test
+ public void testGetDeltaPartitionsWithProjectorWithSameBeginAndEnd() {
+ // begin == end → empty result (open-closed interval: (begin, end])
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ Options opts = new Options();
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_PATTERN, "$date");
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd");
+ CoreOptions options = new CoreOptions(opts);
+
+ BinaryRow partition = row(Lists.newArrayList("EU", "20250720"));
+
+ List<BinaryRow> deltas =
+ ChainTableUtils.getDeltaPartitionsWithProjector(
+ partition, partition, options, chainComparator, projector);
+
+ assertThat(deltas).isEmpty();
+ }
+
+ @Test
+ public void testGetDeltaPartitionsWithProjectorWithConsecutiveDays() {
+ // begin: (EU, 20250720), end: (EU, 20250721) → only (EU, 20250721)
+ RowType fullType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ Options opts = new Options();
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_PATTERN, "$date");
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd");
+ CoreOptions options = new CoreOptions(opts);
+
+ BinaryRow begin = row(Lists.newArrayList("EU", "20250720"));
+ BinaryRow end = row(Lists.newArrayList("EU", "20250721"));
+
+ List<BinaryRow> deltas =
+ ChainTableUtils.getDeltaPartitionsWithProjector(
+ begin, end, options, chainComparator, projector);
+
+ assertThat(deltas).hasSize(1);
+ assertThat(getString(deltas.get(0), 0)).isEqualTo("EU");
+ assertThat(getString(deltas.get(0), 1)).isEqualTo("20250721");
+ }
+
+ @Test
+ public void testGetDeltaPartitionsWithProjectorWithGroupPreserved() {
+ // Verify that group part from beginPartition is preserved for all results
+ RowType fullType =
+ RowType.builder()
+ .field("country", DataTypes.STRING().notNull())
+ .field("city", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ // chain keys: (date), group keys: (country, city)
+ ChainPartitionProjector projector = new ChainPartitionProjector(fullType, 1);
+ RecordComparator chainComparator = (a, b) -> a.getString(0).compareTo(b.getString(0));
+
+ Options opts = new Options();
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_PATTERN, "$date");
+ opts.set(CoreOptions.PARTITION_TIMESTAMP_FORMATTER, "yyyyMMdd");
+ CoreOptions options = new CoreOptions(opts);
+
+ BinaryRow begin = row(Lists.newArrayList("China", "Beijing", "20250720"));
+ BinaryRow end = row(Lists.newArrayList("China", "Beijing", "20250723"));
+
+ List<BinaryRow> deltas =
+ ChainTableUtils.getDeltaPartitionsWithProjector(
+ begin, end, options, chainComparator, projector);
+
+ assertThat(deltas).hasSize(3);
+ for (BinaryRow delta : deltas) {
+ assertThat(getString(delta, 0)).isEqualTo("China");
+ assertThat(getString(delta, 1)).isEqualTo("Beijing");
+ }
+ assertThat(getString(deltas.get(0), 2)).isEqualTo("20250721");
+ assertThat(getString(deltas.get(1), 2)).isEqualTo("20250722");
+ assertThat(getString(deltas.get(2), 2)).isEqualTo("20250723");
+ }
+
+ // ========================== Tests for createGroupChainPredicate ==========================
+
+ @Test
+ public void testCreateGroupChainPredicateWithOneGroupOneChainLessThan() {
+ // partition: (region, date), groupFieldCount=1
+ // Expected: region=equal AND date=lessThan
+ RowType rowType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ BinaryRow partition = row(Lists.newArrayList("US", "20250726"));
+ RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ Predicate predicate =
+ ChainTableUtils.createGroupChainPredicate(
+ partition,
+ converter,
+ 1,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v));
+
+ // Expected: region='US' AND date < '20250726'
+ List<Predicate> expected = new ArrayList<>();
+ expected.add(builder.equal(0, BinaryString.fromString("US")));
+ expected.add(builder.lessThan(1, BinaryString.fromString("20250726")));
+ Predicate expectedPredicate = PredicateBuilder.and(expected);
+
+ assertThat(predicate).isEqualTo(expectedPredicate);
+
+ // Verify: (US, 20250725) should match
+ assertThat(predicate.test(row(Lists.newArrayList("US", "20250725")))).isTrue();
+
+ // Verify: (US, 20250726) should NOT match (not less than)
+ assertThat(predicate.test(partition)).isFalse();
+
+ // Verify: (EU, 20250725) should NOT match (region mismatch)
+ assertThat(predicate.test(row(Lists.newArrayList("EU", "20250725")))).isFalse();
+ }
+
+ @Test
+ public void testCreateGroupChainPredicateWithCombinedWithLinearForLessOrEqual() {
+ // Correct pattern: OR(groupChainPredicate(lessThan), linearPredicate(equal))
+ // to achieve partition <= target with group exact match.
+ // This is the pattern used in planChainInGroup.
+ RowType rowType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ BinaryRow partition = row(Lists.newArrayList("EU", "20250801"));
+ RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ List<Predicate> predicates = new ArrayList<>();
+ predicates.add(
+ ChainTableUtils.createGroupChainPredicate(
+ partition,
+ converter,
+ 1,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v)));
+ predicates.add(
+ ChainTableUtils.createLinearPredicate(
+ partition, converter, (Integer i, Object v) -> builder.equal(i, v)));
+ Predicate combined = PredicateBuilder.or(predicates);
+
+ // Exact match → match (via linearPredicate)
+ assertThat(combined.test(partition)).isTrue();
+
+ // Earlier date, same region → match (via groupChainPredicate)
+ assertThat(combined.test(row(Lists.newArrayList("EU", "20250731")))).isTrue();
+
+ // Later date → no match
+ assertThat(combined.test(row(Lists.newArrayList("EU", "20250802")))).isFalse();
+
+ // Different region → no match
+ assertThat(combined.test(row(Lists.newArrayList("US", "20250801")))).isFalse();
+ }
+
+ @Test
+ public void testCreateGroupChainPredicateWithOneGroupTwoChainLessThan() {
+ // partition: (region, dt, hour), groupFieldCount=1
+ // Expected: region=equal AND triangular(dt, hour) with lessThan
+ // → region='US' AND (dt < '20250810' OR (dt = '20250810' AND hour < '23'))
+ RowType rowType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("dt", DataTypes.STRING().notNull())
+ .field("hour", DataTypes.STRING().notNull())
+ .build();
+
+ BinaryRow partition = row(Lists.newArrayList("US", "20250810", "23"));
+ RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ Predicate predicate =
+ ChainTableUtils.createGroupChainPredicate(
+ partition,
+ converter,
+ 1,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v));
+
+ // (US, 20250809, 23) → dt < 20250810 → match
+ assertThat(predicate.test(row(Lists.newArrayList("US", "20250809", "23")))).isTrue();
+
+ // (US, 20250810, 22) → dt=20250810 AND hour<23 → match
+ assertThat(predicate.test(row(Lists.newArrayList("US", "20250810", "22")))).isTrue();
+
+ // (US, 20250810, 23) → not less than → no match
+ assertThat(predicate.test(partition)).isFalse();
+
+ // (EU, 20250809, 23) → region mismatch → no match
+ assertThat(predicate.test(row(Lists.newArrayList("EU", "20250809", "23")))).isFalse();
+ }
+
+ @Test
+ public void testCreateGroupChainPredicateWithOneGroupTwoChainLessOrEqual() {
+ // partition: (region, dt, hour), groupFieldCount=1
+ // Combine groupChainPredicate(lessThan) OR linearPredicate(equal)
+ // to achieve <= semantics with correct multi-chain-field handling.
+ RowType rowType =
+ RowType.builder()
+ .field("region", DataTypes.STRING().notNull())
+ .field("dt", DataTypes.STRING().notNull())
+ .field("hour", DataTypes.STRING().notNull())
+ .build();
+
+ BinaryRow partition = row(Lists.newArrayList("US", "20250810", "14"));
+ RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ List<Predicate> predicates = new ArrayList<>();
+ predicates.add(
+ ChainTableUtils.createGroupChainPredicate(
+ partition,
+ converter,
+ 1,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v)));
+ predicates.add(
+ ChainTableUtils.createLinearPredicate(
+ partition, converter, (Integer i, Object v) -> builder.equal(i, v)));
+ Predicate combined = PredicateBuilder.or(predicates);
+
+ // (US, 20250810, 14) → exact match → match (via linear)
+ assertThat(combined.test(partition)).isTrue();
+
+ // (US, 20250810, 13) → same dt, hour < 14 → match (via groupChain)
+ assertThat(combined.test(row(Lists.newArrayList("US", "20250810", "13")))).isTrue();
+
+ // (US, 20250809, 23) → dt < 20250810 → match (via groupChain)
+ assertThat(combined.test(row(Lists.newArrayList("US", "20250809", "23")))).isTrue();
+
+ // (US, 20250810, 15) → same dt, hour > 14, no exact match → no match
+ assertThat(combined.test(row(Lists.newArrayList("US", "20250810", "15")))).isFalse();
+
+ // (US, 20250811, 00) → dt > 20250810 → no match
+ assertThat(combined.test(row(Lists.newArrayList("US", "20250811", "00")))).isFalse();
+
+ // (EU, 20250810, 14) → region mismatch → no match
+ assertThat(combined.test(row(Lists.newArrayList("EU", "20250810", "14")))).isFalse();
+ }
+
+ @Test
+ public void testCreateGroupChainPredicateWithZeroGroup() {
+ // groupFieldCount=0 → same as createTriangularPredicate
+ RowType rowType =
+ RowType.builder()
+ .field("dt", DataTypes.STRING().notNull())
+ .field("hour", DataTypes.STRING().notNull())
+ .build();
+
+ BinaryRow partition = row(Lists.newArrayList("20250810", "23"));
+ RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ Predicate groupChainPredicate =
+ ChainTableUtils.createGroupChainPredicate(
+ partition,
+ converter,
+ 0,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v));
+
+ Predicate triangularPredicate =
+ ChainTableUtils.createTriangularPredicate(
+ partition,
+ converter,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v));
+
+ // When groupFieldCount=0, createGroupChainPredicate should produce
+ // the same result as createTriangularPredicate
+ assertThat(groupChainPredicate).isEqualTo(triangularPredicate);
+ }
+
+ @Test
+ public void testCreateGroupChainPredicateWithTwoGroupOneChain() {
+ // partition: (country, city, date), groupFieldCount=2
+ // Expected: country=equal AND city=equal AND date=lessThan
+ RowType rowType =
+ RowType.builder()
+ .field("country", DataTypes.STRING().notNull())
+ .field("city", DataTypes.STRING().notNull())
+ .field("date", DataTypes.STRING().notNull())
+ .build();
+
+ BinaryRow partition = row(Lists.newArrayList("China", "Beijing", "20250801"));
+ RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(rowType);
+ PredicateBuilder builder = new PredicateBuilder(rowType);
+
+ Predicate predicate =
+ ChainTableUtils.createGroupChainPredicate(
+ partition,
+ converter,
+ 2,
+ (Integer i, Object v) -> builder.equal(i, v),
+ (Integer i, Object v) -> builder.lessThan(i, v));
+
+ // (China, Beijing, 20250731) → match
+ assertThat(predicate.test(row(Lists.newArrayList("China", "Beijing", "20250731"))))
+ .isTrue();
+
+ // (China, Beijing, 20250801) → not less than → no match
+ assertThat(predicate.test(partition)).isFalse();
+
+ // (China, Shanghai, 20250731) → city mismatch → no match
+ assertThat(predicate.test(row(Lists.newArrayList("China", "Shanghai", "20250731"))))
+ .isFalse();
+
+ // (US, Beijing, 20250731) → country mismatch → no match
+ assertThat(predicate.test(row(Lists.newArrayList("US", "Beijing", "20250731")))).isFalse();
+ }
}
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
index 83a77f1232..1907c7fcf3 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkChainTableITCase.java
@@ -842,4 +842,1460 @@ public class SparkChainTableITCase {
spark.sql("DROP TABLE IF EXISTS `my_db1`.`chain_test_t`;");
spark.close();
}
+
+ @Test
+ public void testChainTableWithGroupPartition(@TempDir java.nio.file.Path tempDir)
+ throws IOException {
+ Path warehousePath = new Path("file:" + tempDir.toString());
+ SparkSession.Builder builder =
+ SparkSession.builder()
+ .config("spark.sql.warehouse.dir", warehousePath.toString())
+ // with hive metastore
+ .config("spark.sql.catalogImplementation", "hive")
+ .config("hive.metastore.uris", "thrift://localhost:" + PORT)
+ .config("spark.sql.catalog.spark_catalog", SparkCatalog.class.getName())
+ .config("spark.sql.catalog.spark_catalog.metastore", "hive")
+ .config(
+ "spark.sql.catalog.spark_catalog.hive.metastore.uris",
+ "thrift://localhost:" + PORT)
+ .config("spark.sql.catalog.spark_catalog.format-table.enabled", "true")
+ .config(
+ "spark.sql.catalog.spark_catalog.warehouse",
+ warehousePath.toString())
+ .config(
+ "spark.sql.extensions",
+ "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
+ .master("local[2]");
+ SparkSession spark = builder.getOrCreate();
+ spark.sql("CREATE DATABASE IF NOT EXISTS my_db1");
+ spark.sql("USE spark_catalog.my_db1");
+
+ /** Create table */
+ spark.sql(
+ "CREATE TABLE IF NOT EXISTS \n"
+ + " `my_db1`.`chain_test` (\n"
+ + " `t1` BIGINT COMMENT 't1',\n"
+ + " `t2` BIGINT COMMENT 't2',\n"
+ + " `t3` STRING COMMENT 't3'\n"
+ + " ) PARTITIONED BY (`region` STRING, `dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
+ + "WITH\n"
+ + " SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+ + " 'bucket-key' = 't1',\n"
+ + " 'primary-key' = 'region,dt,t1',\n"
+ + " 'partition.timestamp-pattern' = '$dt',\n"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd',\n"
+ + " 'chain-table.enabled' = 'true',\n"
+ + " 'bucket' = '2',\n"
+ + " 'merge-engine' = 'deduplicate', \n"
+ + " 'sequence.field' = 't2',\n"
+ + " 'chain-table.chain-partition-keys' = 'dt'\n"
+ + " )");
+
+ /** Create branch */
+ spark.sql("CALL sys.create_branch('my_db1.chain_test', 'snapshot');");
+ spark.sql("CALL sys.create_branch('my_db1.chain_test', 'delta')");
+
+ /** Set branch */
+ spark.sql(
+ "ALTER TABLE my_db1.chain_test SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot', "
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.sql(
+ "ALTER TABLE `my_db1`.`chain_test$branch_snapshot` SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot',"
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.sql(
+ "ALTER TABLE `my_db1`.`chain_test$branch_delta` SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot',"
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.close();
+ spark = builder.getOrCreate();
+
+ /** Write main branch */
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810') values (11, 1, '1'),(12, 1, '1');");
+
+ /** Write delta branch */
+ spark.sql("set spark.paimon.branch=delta;");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250809') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810') values (1, 2, '1-1' ),(3, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250811') values (2, 2, '1-1' ),(4, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250812') values (3, 2, '1-1' ),(4, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250813') values (5, 1, '1' ),(6, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250814') values (5, 2, '1-1' ),(6, 2, '1-1' );");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250809') values (11, 1, '1'),(12, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810') values (11, 2, '1-1' ),(13, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250811') values (12, 2, '1-1' ),(14, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250812') values (13, 2, '1-1' ),(14, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250813') values (15, 1, '1' ),(16, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250814') values (15, 2, '1-1' ),(16, 2, '1-1' );");
+
+ /** Write snapshot branch */
+ spark.sql("set spark.paimon.branch=snapshot;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810') values (1, 2, '1-1'),(2, 1, '1'),(3, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250812') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250814') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1'), (5, 1, '1' ), (6, 1, '1');");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810') values (11, 2, '1-1'),(12, 1, '1'),(13, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250812') values (11, 2, '1-1'),(12, 2, '1-1'),(13, 2, '1-1'), (14, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250814') values (11, 2, '1-1'),(12, 2, '1-1'),(13, 2, '1-1'), (14, 2, '1-1'), (15, 1, '1' ), (16, 1, '1');");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ /** Main read */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250810'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810]",
+ "[2,1,1,CN,20250810]",
+ "[11,1,1,US,20250810]",
+ "[12,1,1,US,20250810]");
+
+ /** Snapshot read */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250814]",
+ "[2,2,1-1,CN,20250814]",
+ "[3,2,1-1,CN,20250814]",
+ "[4,2,1-1,CN,20250814]",
+ "[5,1,1,CN,20250814]",
+ "[6,1,1,CN,20250814]",
+ "[11,2,1-1,US,20250814]",
+ "[12,2,1-1,US,20250814]",
+ "[13,2,1-1,US,20250814]",
+ "[14,2,1-1,US,20250814]",
+ "[15,1,1,US,20250814]",
+ "[16,1,1,US,20250814]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250814]",
+ "[2,2,1-1,CN,20250814]",
+ "[3,2,1-1,CN,20250814]",
+ "[4,2,1-1,CN,20250814]",
+ "[5,1,1,CN,20250814]",
+ "[6,1,1,CN,20250814]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,20250814]",
+ "[12,2,1-1,US,20250814]",
+ "[13,2,1-1,US,20250814]",
+ "[14,2,1-1,US,20250814]",
+ "[15,1,1,US,20250814]",
+ "[16,1,1,US,20250814]");
+
+ /** Chain read */
+ /** 1. non pre snapshot */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250809]",
+ "[2,1,1,CN,20250809]",
+ "[11,1,1,US,20250809]",
+ "[12,1,1,US,20250809]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[1,1,1,CN,20250809]", "[2,1,1,CN,20250809]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[11,1,1,US,20250809]", "[12,1,1,US,20250809]");
+ /** 2. has pre snapshot */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250811]",
+ "[2,2,1-1,CN,20250811]",
+ "[3,1,1,CN,20250811]",
+ "[4,1,1,CN,20250811]",
+ "[11,2,1-1,US,20250811]",
+ "[12,2,1-1,US,20250811]",
+ "[13,1,1,US,20250811]",
+ "[14,1,1,US,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250811]",
+ "[2,2,1-1,CN,20250811]",
+ "[3,1,1,CN,20250811]",
+ "[4,1,1,CN,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,20250811]",
+ "[12,2,1-1,US,20250811]",
+ "[13,1,1,US,20250811]",
+ "[14,1,1,US,20250811]");
+
+ /** Multi partition Read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810]",
+ "[2,1,1,CN,20250810]",
+ "[1,2,1-1,CN,20250811]",
+ "[2,2,1-1,CN,20250811]",
+ "[3,1,1,CN,20250811]",
+ "[4,1,1,CN,20250811]",
+ "[1,2,1-1,CN,20250812]",
+ "[2,2,1-1,CN,20250812]",
+ "[3,2,1-1,CN,20250812]",
+ "[4,2,1-1,CN,20250812]",
+ "[11,1,1,US,20250810]",
+ "[12,1,1,US,20250810]",
+ "[11,2,1-1,US,20250811]",
+ "[12,2,1-1,US,20250811]",
+ "[13,1,1,US,20250811]",
+ "[14,1,1,US,20250811]",
+ "[11,2,1-1,US,20250812]",
+ "[12,2,1-1,US,20250812]",
+ "[13,2,1-1,US,20250812]",
+ "[14,2,1-1,US,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810]",
+ "[2,1,1,CN,20250810]",
+ "[1,2,1-1,CN,20250811]",
+ "[2,2,1-1,CN,20250811]",
+ "[3,1,1,CN,20250811]",
+ "[4,1,1,CN,20250811]",
+ "[1,2,1-1,CN,20250812]",
+ "[2,2,1-1,CN,20250812]",
+ "[3,2,1-1,CN,20250812]",
+ "[4,2,1-1,CN,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,1,1,US,20250810]",
+ "[12,1,1,US,20250810]",
+ "[11,2,1-1,US,20250811]",
+ "[12,2,1-1,US,20250811]",
+ "[13,1,1,US,20250811]",
+ "[14,1,1,US,20250811]",
+ "[11,2,1-1,US,20250812]",
+ "[12,2,1-1,US,20250812]",
+ "[13,2,1-1,US,20250812]",
+ "[14,2,1-1,US,20250812]");
+
+ /** Incremental read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[2,2,1-1,CN,20250811]",
+ "[4,1,1,CN,20250811]",
+ "[12,2,1-1,US,20250811]",
+ "[14,1,1,US,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[2,2,1-1,CN,20250811]", "[4,1,1,CN,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'US' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[12,2,1-1,US,20250811]", "[14,1,1,US,20250811]");
+
+ /** Multi partition incremental read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810]",
+ "[3,1,1,CN,20250810]",
+ "[2,2,1-1,CN,20250811]",
+ "[4,1,1,CN,20250811]",
+ "[3,2,1-1,CN,20250812]",
+ "[4,2,1-1,CN,20250812]",
+ "[11,2,1-1,US,20250810]",
+ "[13,1,1,US,20250810]",
+ "[12,2,1-1,US,20250811]",
+ "[14,1,1,US,20250811]",
+ "[13,2,1-1,US,20250812]",
+ "[14,2,1-1,US,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810]",
+ "[3,1,1,CN,20250810]",
+ "[2,2,1-1,CN,20250811]",
+ "[4,1,1,CN,20250811]",
+ "[3,2,1-1,CN,20250812]",
+ "[4,2,1-1,CN,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'US' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,20250810]",
+ "[13,1,1,US,20250810]",
+ "[12,2,1-1,US,20250811]",
+ "[14,1,1,US,20250811]",
+ "[13,2,1-1,US,20250812]",
+ "[14,2,1-1,US,20250812]");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ spark.sql("set spark.paimon.branch=delta;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` values (5, 2, '1', 'CN', '20250813'),(6, 2, '1', 'CN', '20250814'), (15, 2, '1', 'US', '20250813'),(16, 2, '1', 'US', '20250814');");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ Dataset<Row> df =
+ spark.sql(
+ "SELECT t1,t2,t3 FROM `my_db1`.`chain_test$branch_snapshot` where dt = '20250814'");
+ assertThat(df.count()).isEqualTo(0);
+ df =
+ spark.sql(
+ "SELECT t1,t2,t3 FROM `my_db1`.`chain_test$branch_delta` where dt = '20250814'");
+ assertThat(df.count()).isEqualTo(2);
+
+ spark.close();
+ spark = builder.getOrCreate();
+ /** Drop table */
+ spark.sql("DROP TABLE IF EXISTS `my_db1`.`chain_test`;");
+
+ spark.close();
+ }
+
+ @Test
+ public void testHourlyChainTableWithGroupPartition(@TempDir java.nio.file.Path tempDir)
+ throws IOException {
+ Path warehousePath = new Path("file:" + tempDir.toString());
+ SparkSession.Builder builder =
+ SparkSession.builder()
+ .config("spark.sql.warehouse.dir", warehousePath.toString())
+ // with hive metastore
+ .config("spark.sql.catalogImplementation", "hive")
+ .config("hive.metastore.uris", "thrift://localhost:" + PORT)
+ .config("spark.sql.catalog.spark_catalog", SparkCatalog.class.getName())
+ .config("spark.sql.catalog.spark_catalog.metastore", "hive")
+ .config(
+ "spark.sql.catalog.spark_catalog.hive.metastore.uris",
+ "thrift://localhost:" + PORT)
+ .config("spark.sql.catalog.spark_catalog.format-table.enabled", "true")
+ .config(
+ "spark.sql.catalog.spark_catalog.warehouse",
+ warehousePath.toString())
+ .config(
+ "spark.sql.extensions",
+ "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
+ .master("local[2]");
+ SparkSession spark = builder.getOrCreate();
+ spark.sql("CREATE DATABASE IF NOT EXISTS my_db1");
+ spark.sql("USE spark_catalog.my_db1");
+
+ /** Create table */
+ spark.sql(
+ "CREATE TABLE IF NOT EXISTS \n"
+ + " `my_db1`.`chain_test` (\n"
+ + " `t1` BIGINT COMMENT 't1',\n"
+ + " `t2` BIGINT COMMENT 't2',\n"
+ + " `t3` STRING COMMENT 't3'\n"
+ + " ) PARTITIONED BY (`region` STRING, `dt` STRING COMMENT 'dt', `hour` STRING COMMENT 'hour') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
+ + "WITH\n"
+ + " SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+ + " 'bucket-key' = 't1',\n"
+ + " 'primary-key' = 'region,dt,hour,t1',\n"
+ + " 'partition.timestamp-pattern' = '$dt $hour:00:00',\n"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd HH:mm:ss',\n"
+ + " 'chain-table.enabled' = 'true',\n"
+ + " 'bucket' = '2',\n"
+ + " 'merge-engine' = 'deduplicate', \n"
+ + " 'sequence.field' = 't2',\n"
+ + " 'chain-table.chain-partition-keys' = 'dt,hour'\n"
+ + " )");
+
+ /** Create branch */
+ spark.sql("CALL sys.create_branch('my_db1.chain_test', 'snapshot');");
+ spark.sql("CALL sys.create_branch('my_db1.chain_test', 'delta')");
+
+ /** Set branch */
+ spark.sql(
+ "ALTER TABLE my_db1.chain_test SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot', "
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.sql(
+ "ALTER TABLE `my_db1`.`chain_test$branch_snapshot` SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot',"
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.sql(
+ "ALTER TABLE `my_db1`.`chain_test$branch_delta` SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot',"
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.close();
+ spark = builder.getOrCreate();
+
+ /** Write main branch */
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810', hour = '22') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810', hour = '22') values (11, 1, '1'),(12, 1, '1');");
+
+ /** Write delta branch */
+ spark.sql("set spark.paimon.branch=delta;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810', hour = '21') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810', hour = '22') values (1, 2, '1-1' ),(3, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810', hour = '23') values (2, 2, '1-1' ),(4, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250811', hour = '00') values (3, 2, '1-1' ),(4, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250811', hour = '01') values (5, 1, '1' ),(6, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250811', hour = '02') values (5, 2, '1-1' ),(6, 2, '1-1' );");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810', hour = '21') values (11, 1, '1'),(12, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810', hour = '22') values (11, 2, '1-1' ),(13, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810', hour = '23') values (12, 2, '1-1' ),(14, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250811', hour = '00') values (13, 2, '1-1' ),(14, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250811', hour = '01') values (15, 1, '1' ),(16, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250811', hour = '02') values (15, 2, '1-1' ),(16, 2, '1-1' );");
+
+ /** Write snapshot branch */
+ spark.sql("set spark.paimon.branch=snapshot;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250810', hour = '22') values (1, 2, '1-1'),(2, 1, '1'),(3, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250811', hour = '00') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', dt = '20250811', hour = '02') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1'), (5, 1, '1' ), (6, 1, '1');");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250810', hour = '22') values (11, 2, '1-1'),(12, 1, '1'),(13, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250811', hour = '00') values (11, 2, '1-1'),(12, 2, '1-1'),(13, 2, '1-1'), (14, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', dt = '20250811', hour = '02') values (11, 2, '1-1'),(12, 2, '1-1'),(13, 2, '1-1'), (14, 2, '1-1'), (15, 1, '1' ), (16, 1, '1');");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ /** Main read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt = '20250810' and hour = '22'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810,22]",
+ "[2,1,1,CN,20250810,22]",
+ "[11,1,1,US,20250810,22]",
+ "[12,1,1,US,20250810,22]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250810' and hour = '22'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[1,1,1,CN,20250810,22]", "[2,1,1,CN,20250810,22]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250810' and hour = '22'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[11,1,1,US,20250810,22]", "[12,1,1,US,20250810,22]");
+
+ /** Snapshot read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt = '20250811' and hour = '02'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250811,02]",
+ "[2,2,1-1,CN,20250811,02]",
+ "[3,2,1-1,CN,20250811,02]",
+ "[4,2,1-1,CN,20250811,02]",
+ "[5,1,1,CN,20250811,02]",
+ "[6,1,1,CN,20250811,02]",
+ "[11,2,1-1,US,20250811,02]",
+ "[12,2,1-1,US,20250811,02]",
+ "[13,2,1-1,US,20250811,02]",
+ "[14,2,1-1,US,20250811,02]",
+ "[15,1,1,US,20250811,02]",
+ "[16,1,1,US,20250811,02]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250811' and hour = '02'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250811,02]",
+ "[2,2,1-1,CN,20250811,02]",
+ "[3,2,1-1,CN,20250811,02]",
+ "[4,2,1-1,CN,20250811,02]",
+ "[5,1,1,CN,20250811,02]",
+ "[6,1,1,CN,20250811,02]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250811' and hour = '02'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,20250811,02]",
+ "[12,2,1-1,US,20250811,02]",
+ "[13,2,1-1,US,20250811,02]",
+ "[14,2,1-1,US,20250811,02]",
+ "[15,1,1,US,20250811,02]",
+ "[16,1,1,US,20250811,02]");
+
+ /** Chain read */
+ /** 1. non pre snapshot */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt = '20250810' and hour = '21'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810,21]",
+ "[2,1,1,CN,20250810,21]",
+ "[11,1,1,US,20250810,21]",
+ "[12,1,1,US,20250810,21]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250810' and hour = '21'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[1,1,1,CN,20250810,21]", "[2,1,1,CN,20250810,21]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250810' and hour = '21'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[11,1,1,US,20250810,21]", "[12,1,1,US,20250810,21]");
+ /** 2. has pre snapshot */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810,23]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[3,1,1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]",
+ "[11,2,1-1,US,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[13,1,1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810,23]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[3,1,1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[13,1,1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ /** Multi partition Read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt = '20250810' and hour in ('22', '23');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810,22]",
+ "[2,1,1,CN,20250810,22]",
+ "[1,2,1-1,CN,20250810,23]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[3,1,1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]",
+ "[11,1,1,US,20250810,22]",
+ "[12,1,1,US,20250810,22]",
+ "[11,2,1-1,US,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[13,1,1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and dt = '20250810' and hour in ('22', '23');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,20250810,22]",
+ "[2,1,1,CN,20250810,22]",
+ "[1,2,1-1,CN,20250810,23]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[3,1,1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250810' and hour in ('22', '23');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,1,1,US,20250810,22]",
+ "[12,1,1,US,20250810,22]",
+ "[11,2,1-1,US,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[13,1,1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ /** Incremental read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[2,2,1-1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[2,2,1-1,CN,20250810,23]", "[4,1,1,CN,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'US' and dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[12,2,1-1,US,20250810,23]", "[14,1,1,US,20250810,23]");
+
+ /** Multi partition incremental read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where dt = '20250810' and hour in ('22', '23');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810,22]",
+ "[3,1,1,CN,20250810,22]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]",
+ "[11,2,1-1,US,20250810,22]",
+ "[13,1,1,US,20250810,22]",
+ "[12,2,1-1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and dt = '20250810' and hour in ('22', '23');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810,22]",
+ "[3,1,1,CN,20250810,22]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'US' and dt = '20250810' and hour in ('22', '23');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,20250810,22]",
+ "[13,1,1,US,20250810,22]",
+ "[12,2,1-1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ /** Hybrid read */
+ assertThat(
+ spark
+ .sql(
+ "select * from `my_db1`.`chain_test` where dt = '20250810' and hour = '23'\n"
+ + "union all\n"
+ + "select * from `my_db1`.`chain_test$branch_delta` where dt = '20250810' and hour = '23'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,20250810,23]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[3,1,1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]",
+ "[2,2,1-1,CN,20250810,23]",
+ "[4,1,1,CN,20250810,23]",
+ "[11,2,1-1,US,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[13,1,1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]",
+ "[12,2,1-1,US,20250810,23]",
+ "[14,1,1,US,20250810,23]");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ spark.sql("set spark.paimon.branch=delta;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` values (6, 2, '1', 'CN', '20250811', '02'), (16, 2, '1', 'US', '20250811', '02');");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ Dataset<Row> df =
+ spark.sql(
+ "SELECT t1,t2,t3 FROM `my_db1`.`chain_test$branch_snapshot` where dt = '20250811' and hour = '02'");
+ assertThat(df.count()).isEqualTo(0);
+ df =
+ spark.sql(
+ "SELECT t1,t2,t3 FROM `my_db1`.`chain_test$branch_delta` where dt = '20250811' and hour = '02'");
+ assertThat(df.count()).isEqualTo(2);
+
+ spark.close();
+ spark = builder.getOrCreate();
+ /** Drop table */
+ spark.sql("DROP TABLE IF EXISTS `my_db1`.`chain_test`;");
+
+ spark.close();
+ }
+
+ @Test
+ public void testChainTableWithMultiGroupPartition(@TempDir java.nio.file.Path tempDir)
+ throws IOException {
+ Path warehousePath = new Path("file:" + tempDir.toString());
+ SparkSession.Builder builder =
+ SparkSession.builder()
+ .config("spark.sql.warehouse.dir", warehousePath.toString())
+ // with hive metastore
+ .config("spark.sql.catalogImplementation", "hive")
+ .config("hive.metastore.uris", "thrift://localhost:" + PORT)
+ .config("spark.sql.catalog.spark_catalog", SparkCatalog.class.getName())
+ .config("spark.sql.catalog.spark_catalog.metastore", "hive")
+ .config(
+ "spark.sql.catalog.spark_catalog.hive.metastore.uris",
+ "thrift://localhost:" + PORT)
+ .config("spark.sql.catalog.spark_catalog.format-table.enabled", "true")
+ .config(
+ "spark.sql.catalog.spark_catalog.warehouse",
+ warehousePath.toString())
+ .config(
+ "spark.sql.extensions",
+ "org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions")
+ .master("local[2]");
+ SparkSession spark = builder.getOrCreate();
+ spark.sql("CREATE DATABASE IF NOT EXISTS my_db1");
+ spark.sql("USE spark_catalog.my_db1");
+
+ /** Create table */
+ spark.sql(
+ "CREATE TABLE IF NOT EXISTS \n"
+ + " `my_db1`.`chain_test` (\n"
+ + " `t1` BIGINT COMMENT 't1',\n"
+ + " `t2` BIGINT COMMENT 't2',\n"
+ + " `t3` STRING COMMENT 't3'\n"
+ + " ) PARTITIONED BY (`region` STRING, `biz_type` STRING COMMENT 'biz_type', `dt` STRING COMMENT 'dt') ROW FORMAT SERDE 'org.apache.paimon.hive.PaimonSerDe'\n"
+ + "WITH\n"
+ + " SERDEPROPERTIES ('serialization.format' = '1') STORED AS INPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonInputFormat' OUTPUTFORMAT 'org.apache.paimon.hive.mapred.PaimonOutputFormat' TBLPROPERTIES (\n"
+ + " 'bucket-key' = 't1',\n"
+ + " 'primary-key' = 'region,biz_type,dt,t1',\n"
+ + " 'partition.timestamp-pattern' = '$dt',\n"
+ + " 'partition.timestamp-formatter' = 'yyyyMMdd',\n"
+ + " 'chain-table.enabled' = 'true',\n"
+ + " 'bucket' = '2',\n"
+ + " 'merge-engine' = 'deduplicate', \n"
+ + " 'sequence.field' = 't2',\n"
+ + " 'chain-table.chain-partition-keys' = 'dt'\n"
+ + " )");
+
+ /** Create branch */
+ spark.sql("CALL sys.create_branch('my_db1.chain_test', 'snapshot');");
+ spark.sql("CALL sys.create_branch('my_db1.chain_test', 'delta')");
+
+ /** Set branch */
+ spark.sql(
+ "ALTER TABLE my_db1.chain_test SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot', "
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.sql(
+ "ALTER TABLE `my_db1`.`chain_test$branch_snapshot` SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot',"
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.sql(
+ "ALTER TABLE `my_db1`.`chain_test$branch_delta` SET tblproperties ("
+ + "'scan.fallback-snapshot-branch' = 'snapshot',"
+ + "'scan.fallback-delta-branch' = 'delta')");
+ spark.close();
+ spark = builder.getOrCreate();
+
+ /** Write main branch */
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250810') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250810') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250810') values (11, 1, '1'),(12, 1, '1');");
+
+ /** Write delta branch */
+ spark.sql("set spark.paimon.branch=delta;");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250809') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250810') values (1, 2, '1-1' ),(3, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250811') values (2, 2, '1-1' ),(4, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250812') values (3, 2, '1-1' ),(4, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250813') values (5, 1, '1' ),(6, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250814') values (5, 2, '1-1' ),(6, 2, '1-1' );");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250809') values (1, 1, '1'),(2, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250810') values (1, 2, '1-1' ),(3, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250811') values (2, 2, '1-1' ),(4, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250812') values (3, 2, '1-1' ),(4, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250813') values (5, 1, '1' ),(6, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250814') values (5, 2, '1-1' ),(6, 2, '1-1' );");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250809') values (11, 1, '1'),(12, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250810') values (11, 2, '1-1' ),(13, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250811') values (12, 2, '1-1' ),(14, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250812') values (13, 2, '1-1' ),(14, 2, '1-1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250813') values (15, 1, '1' ),(16, 1, '1' );");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250814') values (15, 2, '1-1' ),(16, 2, '1-1' );");
+
+ /** Write snapshot branch */
+ spark.sql("set spark.paimon.branch=snapshot;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250810') values (1, 2, '1-1'),(2, 1, '1'),(3, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250812') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '0', dt = '20250814') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1'), (5, 1, '1' ), (6, 1, '1');");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250810') values (1, 2, '1-1'),(2, 1, '1'),(3, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250812') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'CN', biz_type = '1', dt = '20250814') values (1, 2, '1-1'),(2, 2, '1-1'),(3, 2, '1-1'), (4, 2, '1-1'), (5, 1, '1' ), (6, 1, '1');");
+
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250810') values (11, 2, '1-1'),(12, 1, '1'),(13, 1, '1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250812') values (11, 2, '1-1'),(12, 2, '1-1'),(13, 2, '1-1'), (14, 2, '1-1');");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` partition (region = 'US', biz_type = '1', dt = '20250814') values (11, 2, '1-1'),(12, 2, '1-1'),(13, 2, '1-1'), (14, 2, '1-1'), (15, 1, '1' ), (16, 1, '1');");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ /** Main read */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250810'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,0,20250810]",
+ "[2,1,1,CN,0,20250810]",
+ "[1,1,1,CN,1,20250810]",
+ "[2,1,1,CN,1,20250810]",
+ "[11,1,1,US,1,20250810]",
+ "[12,1,1,US,1,20250810]");
+
+ /** Snapshot read */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,0,20250814]",
+ "[2,2,1-1,CN,0,20250814]",
+ "[3,2,1-1,CN,0,20250814]",
+ "[4,2,1-1,CN,0,20250814]",
+ "[5,1,1,CN,0,20250814]",
+ "[6,1,1,CN,0,20250814]",
+ "[1,2,1-1,CN,1,20250814]",
+ "[2,2,1-1,CN,1,20250814]",
+ "[3,2,1-1,CN,1,20250814]",
+ "[4,2,1-1,CN,1,20250814]",
+ "[5,1,1,CN,1,20250814]",
+ "[6,1,1,CN,1,20250814]",
+ "[11,2,1-1,US,1,20250814]",
+ "[12,2,1-1,US,1,20250814]",
+ "[13,2,1-1,US,1,20250814]",
+ "[14,2,1-1,US,1,20250814]",
+ "[15,1,1,US,1,20250814]",
+ "[16,1,1,US,1,20250814]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '0' and dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,0,20250814]",
+ "[2,2,1-1,CN,0,20250814]",
+ "[3,2,1-1,CN,0,20250814]",
+ "[4,2,1-1,CN,0,20250814]",
+ "[5,1,1,CN,0,20250814]",
+ "[6,1,1,CN,0,20250814]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '1' and dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,1,20250814]",
+ "[2,2,1-1,CN,1,20250814]",
+ "[3,2,1-1,CN,1,20250814]",
+ "[4,2,1-1,CN,1,20250814]",
+ "[5,1,1,CN,1,20250814]",
+ "[6,1,1,CN,1,20250814]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250814'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,1,20250814]",
+ "[12,2,1-1,US,1,20250814]",
+ "[13,2,1-1,US,1,20250814]",
+ "[14,2,1-1,US,1,20250814]",
+ "[15,1,1,US,1,20250814]",
+ "[16,1,1,US,1,20250814]");
+
+ /** Chain read */
+ /** 1. non pre snapshot */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,0,20250809]",
+ "[2,1,1,CN,0,20250809]",
+ "[1,1,1,CN,1,20250809]",
+ "[2,1,1,CN,1,20250809]",
+ "[11,1,1,US,1,20250809]",
+ "[12,1,1,US,1,20250809]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '0' and dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[1,1,1,CN,0,20250809]", "[2,1,1,CN,0,20250809]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '1' and dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[1,1,1,CN,1,20250809]", "[2,1,1,CN,1,20250809]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250809'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[11,1,1,US,1,20250809]", "[12,1,1,US,1,20250809]");
+
+ /** 2. has pre snapshot */
+ assertThat(
+ spark.sql("SELECT * FROM `my_db1`.`chain_test` where dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,0,20250811]",
+ "[2,2,1-1,CN,0,20250811]",
+ "[3,1,1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]",
+ "[1,2,1-1,CN,1,20250811]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[3,1,1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]",
+ "[11,2,1-1,US,1,20250811]",
+ "[12,2,1-1,US,1,20250811]",
+ "[13,1,1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '0' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,0,20250811]",
+ "[2,2,1-1,CN,0,20250811]",
+ "[3,1,1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '1' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,1,20250811]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[3,1,1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,1,20250811]",
+ "[12,2,1-1,US,1,20250811]",
+ "[13,1,1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]");
+
+ /** Multi partition Read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,0,20250810]",
+ "[2,1,1,CN,0,20250810]",
+ "[1,2,1-1,CN,0,20250811]",
+ "[2,2,1-1,CN,0,20250811]",
+ "[3,1,1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]",
+ "[1,2,1-1,CN,0,20250812]",
+ "[2,2,1-1,CN,0,20250812]",
+ "[3,2,1-1,CN,0,20250812]",
+ "[4,2,1-1,CN,0,20250812]",
+ "[1,1,1,CN,1,20250810]",
+ "[2,1,1,CN,1,20250810]",
+ "[1,2,1-1,CN,1,20250811]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[3,1,1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]",
+ "[1,2,1-1,CN,1,20250812]",
+ "[2,2,1-1,CN,1,20250812]",
+ "[3,2,1-1,CN,1,20250812]",
+ "[4,2,1-1,CN,1,20250812]",
+ "[11,1,1,US,1,20250810]",
+ "[12,1,1,US,1,20250810]",
+ "[11,2,1-1,US,1,20250811]",
+ "[12,2,1-1,US,1,20250811]",
+ "[13,1,1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]",
+ "[11,2,1-1,US,1,20250812]",
+ "[12,2,1-1,US,1,20250812]",
+ "[13,2,1-1,US,1,20250812]",
+ "[14,2,1-1,US,1,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '0' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,0,20250810]",
+ "[2,1,1,CN,0,20250810]",
+ "[1,2,1-1,CN,0,20250811]",
+ "[2,2,1-1,CN,0,20250811]",
+ "[3,1,1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]",
+ "[1,2,1-1,CN,0,20250812]",
+ "[2,2,1-1,CN,0,20250812]",
+ "[3,2,1-1,CN,0,20250812]",
+ "[4,2,1-1,CN,0,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'CN' and biz_type = '1' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,1,1,CN,1,20250810]",
+ "[2,1,1,CN,1,20250810]",
+ "[1,2,1-1,CN,1,20250811]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[3,1,1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]",
+ "[1,2,1-1,CN,1,20250812]",
+ "[2,2,1-1,CN,1,20250812]",
+ "[3,2,1-1,CN,1,20250812]",
+ "[4,2,1-1,CN,1,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test` where region = 'US' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,1,1,US,1,20250810]",
+ "[12,1,1,US,1,20250810]",
+ "[11,2,1-1,US,1,20250811]",
+ "[12,2,1-1,US,1,20250811]",
+ "[13,1,1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]",
+ "[11,2,1-1,US,1,20250812]",
+ "[12,2,1-1,US,1,20250812]",
+ "[13,2,1-1,US,1,20250812]",
+ "[14,2,1-1,US,1,20250812]");
+
+ /** Incremental read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[2,2,1-1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]",
+ "[12,2,1-1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and biz_type = '0' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[2,2,1-1,CN,0,20250811]", "[4,1,1,CN,0,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and biz_type = '1' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[2,2,1-1,CN,1,20250811]", "[4,1,1,CN,1,20250811]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'US' and dt = '20250811'")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder("[12,2,1-1,US,1,20250811]", "[14,1,1,US,1,20250811]");
+
+ /** Multi partition incremental read */
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,0,20250810]",
+ "[3,1,1,CN,0,20250810]",
+ "[2,2,1-1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]",
+ "[3,2,1-1,CN,0,20250812]",
+ "[4,2,1-1,CN,0,20250812]",
+ "[1,2,1-1,CN,1,20250810]",
+ "[3,1,1,CN,1,20250810]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]",
+ "[3,2,1-1,CN,1,20250812]",
+ "[4,2,1-1,CN,1,20250812]",
+ "[11,2,1-1,US,1,20250810]",
+ "[13,1,1,US,1,20250810]",
+ "[12,2,1-1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]",
+ "[13,2,1-1,US,1,20250812]",
+ "[14,2,1-1,US,1,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and biz_type = '0' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,0,20250810]",
+ "[3,1,1,CN,0,20250810]",
+ "[2,2,1-1,CN,0,20250811]",
+ "[4,1,1,CN,0,20250811]",
+ "[3,2,1-1,CN,0,20250812]",
+ "[4,2,1-1,CN,0,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'CN' and biz_type = '1' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[1,2,1-1,CN,1,20250810]",
+ "[3,1,1,CN,1,20250810]",
+ "[2,2,1-1,CN,1,20250811]",
+ "[4,1,1,CN,1,20250811]",
+ "[3,2,1-1,CN,1,20250812]",
+ "[4,2,1-1,CN,1,20250812]");
+
+ assertThat(
+ spark
+ .sql(
+ "SELECT * FROM `my_db1`.`chain_test$branch_delta` where region = 'US' and dt in ('20250810', '20250811', '20250812');")
+ .collectAsList().stream()
+ .map(Row::toString)
+ .collect(Collectors.toList()))
+ .containsExactlyInAnyOrder(
+ "[11,2,1-1,US,1,20250810]",
+ "[13,1,1,US,1,20250810]",
+ "[12,2,1-1,US,1,20250811]",
+ "[14,1,1,US,1,20250811]",
+ "[13,2,1-1,US,1,20250812]",
+ "[14,2,1-1,US,1,20250812]");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ spark.sql("set spark.paimon.branch=delta;");
+ spark.sql(
+ "insert overwrite table `my_db1`.`chain_test` values (5, 2, '1', 'CN', '0', '20250813'),(6, 2, '1', 'CN', '0', '20250814'), (15, 2, '1', 'US', '1', '20250813'),(16, 2, '1', 'US', '1', '20250814');");
+
+ spark.close();
+ spark = builder.getOrCreate();
+ Dataset<Row> df =
+ spark.sql(
+ "SELECT t1,t2,t3 FROM `my_db1`.`chain_test$branch_snapshot` where dt = '20250814'");
+ assertThat(df.count()).isEqualTo(0);
+ df =
+ spark.sql(
+ "SELECT t1,t2,t3 FROM `my_db1`.`chain_test$branch_delta` where dt = '20250814'");
+ assertThat(df.count()).isEqualTo(2);
+
+ spark.close();
+ spark = builder.getOrCreate();
+ /** Drop table */
+ spark.sql("DROP TABLE IF EXISTS `my_db1`.`chain_test`;");
+
+ spark.close();
+ }
}