This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 0f732af6d3 [core] Support configurable sort order for partial-update
sequence groups (#10547)
0f732af6d3 is described below
commit 0f732af6d3c3ab9efeba95a9d43c134f010dbf88
Author: Ran Tao <[email protected]>
AuthorDate: Wed Oct 14 09:56:13 2026 +0800
[core] Support configurable sort order for partial-update sequence groups
(#10547)
---
.../merge-engine/partial-update.md | 55 +++-
.../compact/PartialUpdateMergeFunction.java | 47 +++-
.../apache/paimon/schema/SchemaManagerUtils.java | 24 +-
.../org/apache/paimon/schema/SchemaValidation.java | 24 +-
.../compact/PartialUpdateMergeFunctionTest.java | 286 +++++++++++++++++++++
.../apache/paimon/schema/SchemaValidationTest.java | 79 +++++-
.../apache/paimon/table/SchemaEvolutionTest.java | 32 +++
.../apache/paimon/flink/PartialUpdateITCase.java | 48 ++++
8 files changed, 575 insertions(+), 20 deletions(-)
diff --git a/docs/docs/primary-key-table/merge-engine/partial-update.md
b/docs/docs/primary-key-table/merge-engine/partial-update.md
index 491b85c221..a71ab8927e 100644
--- a/docs/docs/primary-key-table/merge-engine/partial-update.md
+++ b/docs/docs/primary-key-table/merge-engine/partial-update.md
@@ -105,7 +105,7 @@ SELECT * FROM profiles;
-For fields without aggregation, each group follows these rules:
+For fields without aggregation, each group follows these rules with the
default ascending order:
| Incoming ordering value | Effect on the group |
| --- | --- |
@@ -118,14 +118,60 @@ column cannot be an ordering field or a protected field.
Common ordering types i
`TINYINT`, `SMALLINT`, `INTEGER`, `BIGINT`, `FLOAT`, `DOUBLE`, `DATE`, `TIME`,
`TIMESTAMP`, and
`TIMESTAMP_LTZ`.
+### Sort Order
+
+Set `fields.<ordering-fields>.sequence-group.sort-order` to control each
group's ordering direction:
+
+- `ascending` (default): prefer larger ordering values.
+- `descending`: prefer smaller ordering values.
+
+For example, keep the latest status and the earliest event's channel, even
when updates arrive out of order:
+
+```sql
+CREATE TABLE user_summary (
+ user_id BIGINT PRIMARY KEY NOT ENFORCED,
+ status STRING,
+ status_version BIGINT,
+ channel STRING,
+ first_event_time BIGINT
+) WITH (
+ 'merge-engine' = 'partial-update',
+ 'fields.status_version.sequence-group' = 'status',
+ 'fields.first_event_time.sequence-group' = 'channel',
+ 'fields.first_event_time.sequence-group.sort-order' = 'descending'
+);
+
+INSERT INTO user_summary VALUES (1, 'active', 200, 'organic', 100);
+INSERT INTO user_summary VALUES (1, 'registered', 100, 'referral', 50);
+
+SELECT * FROM user_summary;
+-- 1, active, 200, referral, 50
+```
+
+The status group keeps version `200`, while the event group keeps time `50`
and its channel.
+Equal ordering values still replace non-aggregated protected fields, including
null values.
+Null handling is unchanged: an all-null ordering tuple skips the group, and a
non-null ordering
+value takes precedence over null in either direction.
+
+This option requires a corresponding sequence group on a primary-key table
using `partial-update`.
+Ordering fields must not overlap between groups. Unlike
`sequence.field.sort-order`,
+this option controls an individual group rather than ordering the entire
record.
+
+Choose the direction before writing data. Changing it on an existing table
cannot recover records
+already discarded by earlier merges. All readers, writers, and compaction jobs
must support this
+option before using `descending`; older clients may ignore it and choose the
opposite record.
+
### Multiple Ordering Fields
Use multiple ordering fields when one version is insufficient to break ties.
For example,
`fields.profile_version,profile_offset.sequence-group = name,city` compares
`profile_version`
first, then `profile_offset`. Both ordering columns must be present in the
table schema.
+Set `fields.profile_version,profile_offset.sequence-group.sort-order` to
select one direction
+for the entire tuple; that direction applies to every non-null field
comparison.
The group is skipped only when **all** its ordering fields are null.
Otherwise, the tuple
-participates in comparison, with null sorting before a non-null value:
+participates in comparison, with null sorting before a non-null value in
either direction.
+The following examples use the default ascending order:
| Stored tuple | Incoming tuple | Effect on non-aggregated protected fields |
| --- | --- | --- |
@@ -177,8 +223,9 @@ SELECT * FROM profile_totals;
-- 1, Bea, 11, 8, 100
```
-The stored ordering tuple advances on newer or equal versions. With `sum`, an
older or equal
-version still adds its contribution: a sequence group does **not** deduplicate
repeated events.
+The stored ordering tuple advances on equal or higher-precedence values under
the configured
+sort order. With `sum`, lower-precedence or equal values still contribute: a
sequence group
+does **not** deduplicate repeated events.
The same rules apply to groups with multiple ordering fields.
:::warning
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
index f40bcde8a3..4fc4f9b731 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunction.java
@@ -19,6 +19,7 @@
package org.apache.paimon.mergetree.compact;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.SortOrder;
import org.apache.paimon.KeyValue;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
@@ -55,6 +56,7 @@ import static org.apache.paimon.CoreOptions.FIELDS_PREFIX;
import static org.apache.paimon.CoreOptions.FIELDS_SEPARATOR;
import static
org.apache.paimon.CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_DELETE;
import static
org.apache.paimon.CoreOptions.PARTIAL_UPDATE_REMOVE_RECORD_ON_SEQUENCE_GROUP;
+import static org.apache.paimon.options.ConfigOptions.key;
import static org.apache.paimon.utils.InternalRowUtils.createFieldGetters;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -65,6 +67,7 @@ import static
org.apache.paimon.utils.Preconditions.checkArgument;
public class PartialUpdateMergeFunction implements MergeFunction<KeyValue> {
public static final String SEQUENCE_GROUP = "sequence-group";
+ public static final String SEQUENCE_GROUP_SORT_ORDER = SEQUENCE_GROUP +
".sort-order";
private static final String SEQUENCE_GROUP_PK_ERROR =
"The sequence-group '%s' contains primary key field '%s', "
+ "which is not allowed. Primary key columns cannot be put
in sequence-group.";
@@ -89,6 +92,38 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
.split(FIELDS_SEPARATOR));
}
+ public static boolean isSequenceGroupSortOrderOption(String optionKey) {
+ return optionKey.startsWith(FIELDS_PREFIX + ".")
+ && optionKey.endsWith("." + SEQUENCE_GROUP_SORT_ORDER);
+ }
+
+ /** Parses sequence-group sort orders and validates their configuration. */
+ public static Map<String, SortOrder> parseSequenceGroupSortOrders(Options
options) {
+ Map<String, SortOrder> sequenceGroupSortOrders = new HashMap<>();
+ for (String optionKey : options.toMap().keySet()) {
+ if (isSequenceGroupOptionCandidate(optionKey)) {
+ SortOrder order =
+ options.get(
+ key(optionKey + ".sort-order")
+ .enumType(SortOrder.class)
+ .defaultValue(SortOrder.ASCENDING));
+ sequenceGroupSortOrders.put(optionKey, order);
+ }
+ }
+ for (String optionKey : options.toMap().keySet()) {
+ if (isSequenceGroupSortOrderOption(optionKey)) {
+ String groupKey =
+ optionKey.substring(0, optionKey.length() -
".sort-order".length());
+ checkArgument(
+ sequenceGroupSortOrders.containsKey(groupKey),
+ "Can not set '%s' without defining corresponding
sequence group '%s'.",
+ optionKey,
+ groupKey);
+ }
+ }
+ return sequenceGroupSortOrders;
+ }
+
public static List<String> sequenceGroupProtectedFields(String
optionValue) {
return Arrays.asList(optionValue.split(FIELDS_SEPARATOR));
}
@@ -437,6 +472,7 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
this.fieldSeqComparators = new HashMap<>();
Map<String, Integer> sequenceGroupMap = new HashMap<>();
Map<String, String> sequenceGroupByOrderingField = new HashMap<>();
+ Map<String, SortOrder> sequenceGroupSortOrders =
parseSequenceGroupSortOrders(options);
List<String> allSequenceFields = new ArrayList<>();
List<String> fieldsProtectedBySequenceGroup = new ArrayList<>();
for (Map.Entry<String, String> entry : options.toMap().entrySet())
{
@@ -461,8 +497,13 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
k);
}
+ SortOrder sortOrder = sequenceGroupSortOrders.get(k);
Supplier<FieldsComparator> userDefinedSeqComparator =
- () -> UserDefinedSeqComparator.create(rowType,
sequenceFields, true);
+ () ->
+ UserDefinedSeqComparator.create(
+ rowType,
+ sequenceFields,
+ sortOrder == SortOrder.ASCENDING);
sequenceGroupProtectedFields(v).stream()
.map(fieldName -> requireField(fieldName,
fieldNames))
.forEach(
@@ -593,7 +634,9 @@ public class PartialUpdateMergeFunction implements
MergeFunction<KeyValue> {
projectedSeqComparators.put(
newField,
UserDefinedSeqComparator.create(
- readType, newSequenceFields,
true));
+ readType,
+ newSequenceFields,
+
comparator.isAscendingOrder()));
}
});
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java
index a71451c865..52adcf42b3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManagerUtils.java
@@ -91,6 +91,7 @@ import static
org.apache.paimon.catalog.AbstractCatalog.DB_SUFFIX;
import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
import static org.apache.paimon.catalog.Identifier.UNKNOWN_DATABASE;
import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.SEQUENCE_GROUP;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.SEQUENCE_GROUP_SORT_ORDER;
import static
org.apache.paimon.schema.ColumnDirectiveUtils.applyAddColumnDirective;
import static
org.apache.paimon.schema.ColumnDirectiveUtils.parseAddColumnComment;
import static org.apache.paimon.types.BlobType.isBlobFileField;
@@ -673,11 +674,15 @@ final class SchemaManagerUtils {
}
}
- // case 3: both option key and option value may contain field names
+ // case 3: option keys may contain multiple field names, as may their
values
for (String key : options.keySet()) {
if (key.startsWith(FIELDS_PREFIX)) {
String matchedSuffix = null;
- if (key.endsWith(SEQUENCE_GROUP)) {
+ boolean valueContainsFieldNames = true;
+ if (key.endsWith(SEQUENCE_GROUP_SORT_ORDER)) {
+ matchedSuffix = SEQUENCE_GROUP_SORT_ORDER;
+ valueContainsFieldNames = false;
+ } else if (key.endsWith(SEQUENCE_GROUP)) {
matchedSuffix = SEQUENCE_GROUP;
} else if (key.endsWith(NESTED_KEY)) {
matchedSuffix = NESTED_KEY;
@@ -695,17 +700,22 @@ final class SchemaManagerUtils {
List<String> newKeyFields =
applyNotNestedColumnRename(keyFields,
renameMappings);
- String valueFieldsStr = newOptions.remove(key);
- List<String> valueFields =
Arrays.asList(valueFieldsStr.split(","));
- List<String> newValueFields =
- applyNotNestedColumnRename(valueFields,
renameMappings);
+ String newValue = newOptions.remove(key);
+ if (valueContainsFieldNames) {
+ newValue =
+ String.join(
+ ",",
+ applyNotNestedColumnRename(
+
Arrays.asList(newValue.split(",")),
+ renameMappings));
+ }
newOptions.put(
FIELDS_PREFIX
+ "."
+ String.join(",", newKeyFields)
+ "."
+ matchedSuffix,
- String.join(",", newValueFields));
+ newValue);
}
}
}
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 cd5a3fafed..ab73ca53f2 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
@@ -104,6 +104,8 @@ import static
org.apache.paimon.CoreOptions.STREAMING_READ_OVERWRITE;
import static org.apache.paimon.format.FileFormat.vectorFileFormat;
import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.isSequenceGroupOption;
import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.isSequenceGroupOptionCandidate;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.isSequenceGroupSortOrderOption;
+import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.parseSequenceGroupSortOrders;
import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.sequenceGroupOrderingFields;
import static
org.apache.paimon.mergetree.compact.PartialUpdateMergeFunction.sequenceGroupProtectedFields;
import static org.apache.paimon.schema.TableSchema.PAIMON_07_VERSION;
@@ -192,6 +194,8 @@ public class SchemaValidation {
validateSequenceGroupOrderingFields(schema, options);
+ validateSequenceGroupSortOrder(schema, options);
+
validateMergeFunction(schema);
ChangelogProducer changelogProducer = options.changelogProducer();
@@ -942,8 +946,13 @@ public class SchemaValidation {
// PAIMON-6471: dot paths into a ROW field's
members
// (e.g. fields.row.inner.aggregate-function)
are silently
// dropped today and produce wrong results.
This check
- // rejects them.
- if (segments.length > 3 && fields.length == 1)
{
+ // rejects them, except for the top-level
sequence-group.sort-order
+ // option, whose suffix itself contains a dot.
+ boolean sequenceGroupSortOrder =
+ segments.length == 4 &&
isSequenceGroupSortOrderOption(k);
+ if (segments.length > 3
+ && fields.length == 1
+ && !sequenceGroupSortOrder) {
String parent = fields[0];
schema.fields().stream()
.filter(f ->
f.name().equals(parent))
@@ -960,6 +969,17 @@ public class SchemaValidation {
});
}
+ private static void validateSequenceGroupSortOrder(TableSchema schema,
CoreOptions options) {
+ if (options.toMap().keySet().stream()
+ .anyMatch(optionKey ->
isSequenceGroupSortOrderOption(optionKey))) {
+ checkArgument(
+ !schema.primaryKeys().isEmpty()
+ && options.mergeEngine() ==
MergeEngine.PARTIAL_UPDATE,
+ "Sequence group sort-order options are only supported by
primary-key tables with the partial-update merge engine.");
+ parseSequenceGroupSortOrders(options.toConfiguration());
+ }
+ }
+
private static void validateMergeFunction(TableSchema schema) {
if (schema.primaryKeys().isEmpty()) {
return;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
index d8c70797a5..e00449838e 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/PartialUpdateMergeFunctionTest.java
@@ -26,6 +26,7 @@ import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InternalRowUtils;
import org.apache.paimon.utils.ProjectedRow;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
@@ -101,6 +102,291 @@ public class PartialUpdateMergeFunctionTest {
validate(func, 1, null, null, 6, null, null, 6);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSequenceGroupSortOrder(boolean explicitAscendingOrder) {
+ Options options = new Options();
+ options.set("fields.f2.sequence-group", "f1");
+ if (explicitAscendingOrder) {
+ options.set("fields.f2.sequence-group.sort-order", "ascending");
+ }
+ options.set("fields.f4.sequence-group", "f3");
+ options.set("fields.f4.sequence-group.sort-order", "descending");
+ RowType rowType =
+ RowType.of(
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.INT());
+ MergeFunction<KeyValue> func =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("f0"))
+ .create();
+ func.reset();
+ add(func, 1, 100, 10, 200, 10);
+ add(func, 1, 50, 5, 50, 5);
+ add(func, 1, 70, 7, 70, 7);
+ validate(func, 1, 100, 10, 50, 5);
+ // Equal ascending values are accepted while a null descending value
skips its group.
+ add(func, 1, 101, 10, 999, null);
+ validate(func, 1, 101, 10, 50, 5);
+ // An equal ascending value also replaces a protected field with null.
+ add(func, 1, null, 10, 999, null);
+ validate(func, 1, null, 10, 50, 5);
+ // A higher-precedence ascending value is accepted while a null
descending value skips.
+ add(func, 1, 110, 11, 999, null);
+ validate(func, 1, 110, 11, 50, 5);
+ // A null ascending value skips its group while an equal descending
value is accepted.
+ add(func, 1, 999, null, 30, 5);
+ validate(func, 1, 110, 11, 30, 5);
+ // An equal descending value also replaces a protected field with null.
+ add(func, 1, 999, null, null, 5);
+ validate(func, 1, 110, 11, null, 5);
+ // A null ascending value skips its group while a smaller descending
value is accepted.
+ add(func, 1, 999, null, 31, 4);
+ validate(func, 1, 110, 11, 31, 4);
+ // Null ordering values in both groups leave both groups unchanged.
+ add(func, 1, 999, null, 999, null);
+ validate(func, 1, 110, 11, 31, 4);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "ascending", "descending"})
+ public void testRowSequenceGroupSortOrder(String sortOrder) {
+ Options options = new Options();
+ options.set("fields.f2.sequence-group", "f1");
+ if (!sortOrder.isEmpty()) {
+ options.set("fields.f2.sequence-group.sort-order", sortOrder);
+ }
+ RowType rowType =
+ RowType.of(
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.ROW(DataTypes.FIELD(3, "version",
DataTypes.INT())));
+ MergeFunction<KeyValue> func =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("f0"))
+ .create();
+ GenericRow nullVersion = GenericRow.of((Object) null);
+ boolean ascending = sortOrder.isEmpty() ||
"ascending".equals(sortOrder);
+ GenericRow winningSequence = ascending ? GenericRow.of(20) :
nullVersion;
+ GenericRow[] records = {
+ GenericRow.of(1, 100, null),
+ GenericRow.of(1, 10, GenericRow.of(10)),
+ GenericRow.of(1, 5, GenericRow.of(5)),
+ GenericRow.of(1, 99, nullVersion),
+ GenericRow.of(1, 20, GenericRow.of(20)),
+ GenericRow.of(1, 200, null),
+ GenericRow.of(1, 21, winningSequence),
+ GenericRow.of(1, null, winningSequence)
+ };
+ // A top-level null skips the group. ROW(NULL) participates in
comparison and is smaller
+ // than ROW(5), so descending order prefers it. Equal ROW values still
replace the payload,
+ // including with null.
+ GenericRow[] expected = {
+ GenericRow.of(1, null, null),
+ GenericRow.of(1, 10, GenericRow.of(10)),
+ ascending
+ ? GenericRow.of(1, 10, GenericRow.of(10))
+ : GenericRow.of(1, 5, GenericRow.of(5)),
+ ascending ? GenericRow.of(1, 10, GenericRow.of(10)) :
GenericRow.of(1, 99, nullVersion),
+ ascending ? GenericRow.of(1, 20, GenericRow.of(20)) :
GenericRow.of(1, 99, nullVersion),
+ ascending ? GenericRow.of(1, 20, GenericRow.of(20)) :
GenericRow.of(1, 99, nullVersion),
+ GenericRow.of(1, 21, winningSequence),
+ GenericRow.of(1, null, winningSequence)
+ };
+ func.reset();
+ for (int i = 0; i < records.length; i++) {
+ func.add(
+ new KeyValue()
+ .replace(GenericRow.of(1), sequence++,
RowKind.INSERT, records[i]));
+ assertThat(func.getResult().value())
+ .as("Sort order: %s, input: %s", sortOrder, records[i])
+ .isEqualTo(expected[i]);
+ }
+ }
+
+ @Test
+ public void testDescendingMultiFieldSequenceGroup() {
+ Options options = new Options();
+ options.set("fields.f2,f3.sequence-group", "f1");
+ options.set("fields.f2,f3.sequence-group.sort-order", "descending");
+ RowType rowType =
+ RowType.of(DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.INT());
+ MergeFunction<KeyValue> func =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("f0"))
+ .create();
+ func.reset();
+ add(func, 1, 100, 10, 10);
+ add(func, 1, 200, 10, 5);
+ add(func, 1, 300, 11, 0);
+ validate(func, 1, 200, 10, 5);
+ // With equal first ordering values, a null second value has lower
precedence.
+ add(func, 1, 400, 10, null);
+ validate(func, 1, 200, 10, 5);
+ // In descending order, a smaller first value has higher precedence,
even when the second
+ // value is null.
+ add(func, 1, 600, 9, null);
+ validate(func, 1, 600, 9, null);
+ // An all-null ordering tuple skips the group.
+ add(func, 1, 500, null, null);
+ validate(func, 1, 600, 9, null);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testDescendingSequenceGroupProjectionAndDelete(boolean
project) {
+ Options options = new Options();
+ options.set("fields.f2.sequence-group", "f1");
+ options.set("fields.f2.sequence-group.sort-order", "descending");
+ RowType rowType =
+ RowType.of(DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.INT());
+ MergeFunctionFactory<KeyValue> factory =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("f0"));
+ // Only request payload and key: adjustReadType must restore the
ordering field.
+ RowType readType = project ?
factory.adjustReadType(rowType.project("f1", "f0")) : rowType;
+ MergeFunction<KeyValue> func = project ? factory.create(readType) :
factory.create();
+ ProjectedRow projection = ProjectedRow.from(readType, rowType);
+ GenericRow[] records = {
+ GenericRow.of(1, 100, 10, 0), GenericRow.of(1, 50, 5, 0),
+ GenericRow.of(1, 99, 7, 0), GenericRow.of(1, 99, 4, 0)
+ };
+ RowKind[] kinds = {RowKind.INSERT, RowKind.INSERT, RowKind.DELETE,
RowKind.DELETE};
+ GenericRow[] expected = {
+ GenericRow.of(1, 100, 10, 0), GenericRow.of(1, 50, 5, 0),
+ GenericRow.of(1, 50, 5, 0), GenericRow.of(1, null, 4, 0)
+ };
+ func.reset();
+ for (int i = 0; i < records.length; i++) {
+ func.add(
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ sequence++,
+ kinds[i],
+ projection.replaceRow(records[i])));
+ assertThat(
+ InternalRowUtils.equals(
+ func.getResult().value(),
+ projection.replaceRow(expected[i]),
+ readType))
+ .as("Projected: %s, input: %s, expected: %s", project,
records[i], expected[i])
+ .isTrue();
+ }
+ }
+
+ @Test
+ public void testDescendingSequenceGroupAggregates() {
+ Options options = new Options();
+ options.set("fields.f3.sequence-group", "f1,f2");
+ options.set("fields.f3.sequence-group.sort-order", "descending");
+ options.set("fields.f1.aggregate-function", "last_non_null_value");
+ options.set("fields.f2.aggregate-function", "sum");
+ RowType rowType =
+ RowType.of(DataTypes.INT(), DataTypes.INT(), DataTypes.INT(),
DataTypes.INT());
+ MergeFunction<KeyValue> func =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("f0"))
+ .create();
+ func.reset();
+ // In descending order, sequence 5 has higher precedence than 10; sum
adds both values.
+ add(func, 1, 10, 10, 10);
+ add(func, 1, 5, 5, 5);
+ validate(func, 1, 5, 15, 5);
+ // Sequence 7 has lower precedence than 5: last_non_null retains 5,
while sum adds 7.
+ add(func, 1, 7, 7, 7);
+ validate(func, 1, 5, 22, 5);
+ // Sequence 3 has higher precedence than 5; last_non_null ignores null
while sum adds 3.
+ add(func, 1, null, 3, 3);
+ validate(func, 1, 5, 25, 3);
+ // A null ordering value skips the entire group, including both
aggregate contributions.
+ add(func, 1, 999, 100, null);
+ validate(func, 1, 5, 25, 3);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "value,id"})
+ public void testDescendingSequenceGroupRowDeleteWithProjection(String
selectedFields) {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("value", DataTypes.INT())
+ .field("sequence", DataTypes.INT())
+ .build();
+ Options options = new Options();
+ options.set("fields.sequence.sequence-group", "value");
+ options.set("fields.sequence.sequence-group.sort-order", "descending");
+ options.set("partial-update.remove-record-on-sequence-group",
"sequence");
+ MergeFunctionFactory<KeyValue> factory =
+ PartialUpdateMergeFunction.factory(options, rowType,
ImmutableList.of("id"));
+ RowType readType =
+ factory.adjustReadType(
+ rowType.project(
+ selectedFields.isEmpty()
+ ? new String[0]
+ : selectedFields.split(",")));
+ MergeFunction<KeyValue> function = factory.create(readType);
+ GenericRow[] records = {
+ GenericRow.of(1, 100, 10),
+ GenericRow.of(1, 99, 11),
+ GenericRow.of(1, 99, 5),
+ GenericRow.of(1, 70, 7),
+ GenericRow.of(1, 40, 4)
+ };
+ RowKind[] kinds = {
+ RowKind.INSERT, RowKind.DELETE, RowKind.DELETE, RowKind.INSERT,
RowKind.UPDATE_AFTER
+ };
+ RowKind[] expected = {
+ RowKind.INSERT, RowKind.INSERT, RowKind.DELETE, RowKind.DELETE,
RowKind.INSERT
+ };
+ GenericRow[] expectedProjectedValues = {
+ GenericRow.of(100, 1, 10),
+ GenericRow.of(100, 1, 10),
+ GenericRow.of(99, 1, 5),
+ GenericRow.of(99, 1, 5),
+ GenericRow.of(40, 1, 4)
+ };
+ function.reset();
+ for (int i = 0; i < records.length; i++) {
+ function.add(
+ new KeyValue()
+ .replace(
+ GenericRow.of(1),
+ sequence++,
+ kinds[i],
+ ProjectedRow.from(readType,
rowType).replaceRow(records[i])));
+ assertThat(function.getResult().valueKind())
+ .as("Projection %s, input %s", selectedFields, i)
+ .isEqualTo(expected[i]);
+ if (!selectedFields.isEmpty()) {
+ assertThat(function.getResult().value())
+ .as("Projection %s, input %s", selectedFields, i)
+ .isEqualTo(expectedProjectedValues[i]);
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSequenceGroupOrderingFieldsCannotOverlap(boolean
descending) {
+ Options options = new Options();
+ options.set("fields.f3.sequence-group", "f1");
+ options.set("fields.f3,f4.sequence-group", "f2");
+ if (descending) {
+ options.set("fields.f3,f4.sequence-group.sort-order",
"descending");
+ }
+ RowType rowType =
+ RowType.of(
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.INT());
+ assertThatThrownBy(
+ () ->
+ PartialUpdateMergeFunction.factory(
+ options, rowType,
ImmutableList.of("f0")))
+ .hasMessageContaining("Ordering fields must not overlap
between sequence groups.");
+ }
+
@Test
public void testSequenceGroupWithBlobField() {
Options options = new Options();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index 8c4ac570af..50bf9e8006 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -28,6 +28,7 @@ import org.apache.paimon.types.DataTypes;
import org.assertj.core.api.ThrowableAssert;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.time.LocalDateTime;
@@ -74,6 +75,70 @@ class SchemaValidationTest {
"Option 'primary-key.nullable' can only be enabled for
a table with primary keys.");
}
+ @Test
+ public void testSequenceGroupSortOrderValidation() {
+ Map<String, String> options = new HashMap<>();
+ options.put("merge-engine", "partial-update");
+ options.put("fields.f2.sequence-group", "f3");
+ options.put("fields.f2.sequence-group.sort-order", "descending");
+ assertThatCode(() ->
validateTableSchemaExec(options)).doesNotThrowAnyException();
+ options.put("fields.f2.sequence-group.sort-order", "invalid");
+ assertThatThrownBy(() -> validateTableSchemaExec(options))
+ .hasMessageContaining("Could not parse value 'invalid'");
+ options.put("fields.f2.sequence-group.sort-order", "descending");
+ options.remove("fields.f2.sequence-group");
+ assertThatThrownBy(() -> validateTableSchemaExec(options))
+ .hasMessageContaining("without defining corresponding sequence
group");
+ options.put("fields.f2.sequence-group", "f3");
+ options.put("merge-engine", "deduplicate");
+ assertThatThrownBy(() -> validateTableSchemaExec(options))
+ .hasMessageContaining(
+ "only supported by primary-key tables with the
partial-update merge engine");
+ options.put("merge-engine", "partial-update");
+ TableSchema appendSchema =
+ new TableSchema(
+ 1,
+ Arrays.asList(
+ new DataField(0, "f2", DataTypes.INT()),
+ new DataField(1, "f3", DataTypes.STRING())),
+ 1,
+ emptyList(),
+ emptyList(),
+ options,
+ "");
+ assertThatThrownBy(() -> validateTableSchema(appendSchema))
+ .hasMessageContaining(
+ "only supported by primary-key tables with the
partial-update merge engine");
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"", "ascending", "descending"})
+ public void testRowSequenceGroupSortOrder(String sortOrder) {
+ Map<String, String> options = new HashMap<>();
+ options.put("merge-engine", "partial-update");
+ options.put("fields.seq.sequence-group", "payload");
+ if (!sortOrder.isEmpty()) {
+ options.put("fields.seq.sequence-group.sort-order", sortOrder);
+ }
+ TableSchema schema =
+ new TableSchema(
+ 1,
+ Arrays.asList(
+ new DataField(0, "id",
DataTypes.INT().notNull()),
+ new DataField(1, "payload",
DataTypes.STRING()),
+ new DataField(
+ 2,
+ "seq",
+ DataTypes.ROW(
+ DataTypes.FIELD(3, "version",
DataTypes.INT())))),
+ 3,
+ emptyList(),
+ singletonList("id"),
+ options,
+ "");
+ assertThatCode(() ->
validateTableSchema(schema)).doesNotThrowAnyException();
+ }
+
private void validateTableSchemaExec(Map<String, String> options) {
List<DataField> fields =
Arrays.asList(
@@ -904,8 +969,12 @@ class SchemaValidationTest {
assertThatCode(() ->
validateTableSchemaExec(options)).doesNotThrowAnyException();
}
- @Test
- public void testNestedRowAggregateOptionRejected() {
+ @ParameterizedTest
+ @CsvSource({
+ "aggregation, fields.data.num.aggregate-function, sum",
+ "partial-update, fields.data.num.sequence-group.sort-order, descending"
+ })
+ public void testNestedRowOptionRejected(String mergeEngine, String
optionKey, String value) {
// PAIMON-6471: nested-path agg config was silently ignored, leaving
the parent ROW
// on the default aggregator and producing wrong results without any
error.
List<DataField> fields =
@@ -919,8 +988,8 @@ class SchemaValidationTest {
DataTypes.FIELD(3, "info",
DataTypes.STRING()))));
Map<String, String> options = new HashMap<>();
- options.put("merge-engine", "aggregation");
- options.put("fields.data.num.aggregate-function", "sum");
+ options.put("merge-engine", mergeEngine);
+ options.put(optionKey, value);
options.put("fields.data.sequence-group", "data");
options.put(BUCKET.key(), "1");
@@ -929,7 +998,7 @@ class SchemaValidationTest {
assertThatThrownBy(() -> validateTableSchema(schema))
.hasMessageContaining("Nested-field path is not supported on
ROW field 'data'")
- .hasMessageContaining("fields.data.num.aggregate-function");
+ .hasMessageContaining(optionKey);
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
index ffffc2be01..2e2adc51a2 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/SchemaEvolutionTest.java
@@ -1159,6 +1159,38 @@ public class SchemaEvolutionTest {
assertThat(newSchema.options()).contains(entries);
}
+ @Test
+ public void testRenameSequenceGroupSortOrder() throws Exception {
+ Schema schema =
+ new Schema(
+ ImmutableList.of(
+ new DataField(0, "id", DataTypes.INT()),
+ new DataField(1, "s1", DataTypes.INT()),
+ new DataField(2, "s2", DataTypes.INT()),
+ new DataField(3, "descending",
DataTypes.STRING())),
+ ImmutableList.of(),
+ ImmutableList.of("id"),
+ ImmutableMap.of(
+ "merge-engine",
+ "partial-update",
+ "fields.s1,s2.sequence-group",
+ "descending",
+ "fields.s1,s2.sequence-group.sort-order",
+ "descending"),
+ "");
+ schemaManager.createTable(schema);
+ TableSchema renamed =
+ schemaManager.commitChanges(
+ SchemaChange.renameColumn("s1", "new_s1"),
+ SchemaChange.renameColumn("s2", "new_s2"),
+ SchemaChange.renameColumn("descending", "payload"));
+ assertThat(renamed.options())
+ .doesNotContainKeys(
+ "fields.s1,s2.sequence-group",
"fields.s1,s2.sequence-group.sort-order")
+ .containsEntry("fields.new_s1,new_s2.sequence-group",
"payload")
+
.containsEntry("fields.new_s1,new_s2.sequence-group.sort-order", "descending");
+ }
+
@Test
public void testRenameSeqGroupFields() throws Exception {
ImmutableMap.Builder<String, String> mapBuilder =
ImmutableMap.builder();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
index 681821f190..8422c48d3b 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PartialUpdateITCase.java
@@ -216,6 +216,54 @@ public class PartialUpdateITCase extends CatalogITCaseBase
{
assertThat(sql("SELECT c, d FROM
SG")).containsExactlyInAnyOrder(Row.of(5, null));
}
+ @Test
+ public void testSequenceGroupSortOrder() {
+ sql(
+ "CREATE TABLE user_summary (user_id BIGINT PRIMARY KEY NOT
ENFORCED, "
+ + "status STRING, status_version BIGINT, channel
STRING, first_event_time BIGINT) WITH ("
+ + "'merge-engine'='partial-update', 'bucket'='1',
'write-only'='true', "
+ + "'fields.status_version.sequence-group'='status', "
+ + "'fields.first_event_time.sequence-group'='channel',
"
+ +
"'fields.first_event_time.sequence-group.sort-order'='descending')");
+ sql("INSERT INTO user_summary VALUES (1, 'active', 200, 'organic',
100)");
+ sql("INSERT INTO user_summary VALUES (1, 'registered', 100,
'referral', 50)");
+ assertThat(sql("SELECT * FROM user_summary"))
+ .containsExactly(Row.of(1L, "active", 200L, "referral", 50L));
+ assertThat(sql("SELECT channel, user_id FROM user_summary"))
+ .containsExactly(Row.of("referral", 1L));
+
+ sql("ALTER TABLE user_summary RENAME first_event_time TO event_time");
+ sql("INSERT INTO user_summary VALUES (1, 'newest', 300, 'later', 75)");
+ assertThat(sql("SELECT * FROM user_summary"))
+ .containsExactly(Row.of(1L, "newest", 300L, "referral", 50L));
+
+ tEnv.getConfig().set(TableConfigOptions.TABLE_DML_SYNC, true);
+ sql("CALL sys.compact(`table` => 'default.user_summary',
compact_strategy => 'full')");
+ assertThat(sql("SELECT * FROM user_summary"))
+ .containsExactly(Row.of(1L, "newest", 300L, "referral", 50L));
+ sql("INSERT INTO user_summary VALUES (1, 'old', 100, 'earliest', 25)");
+ assertThat(sql("SELECT channel, status FROM user_summary"))
+ .containsExactly(Row.of("earliest", "newest"));
+
+ sql(
+ "INSERT INTO user_summary VALUES (1, 'ignored', CAST(NULL AS
BIGINT), "
+ + "'earlier', 20)");
+ assertThat(sql("SELECT status, channel FROM user_summary"))
+ .containsExactly(Row.of("newest", "earlier"));
+
+ sql(
+ "INSERT INTO user_summary VALUES (1, 'newest2', 400,
'ignored', "
+ + "CAST(NULL AS BIGINT))");
+ assertThat(sql("SELECT status, channel FROM user_summary"))
+ .containsExactly(Row.of("newest2", "earlier"));
+
+ sql(
+ "INSERT INTO user_summary VALUES (1, 'ignored', CAST(NULL AS
BIGINT), "
+ + "'ignored', CAST(NULL AS BIGINT))");
+ assertThat(sql("SELECT status, channel FROM user_summary"))
+ .containsExactly(Row.of("newest2", "earlier"));
+ }
+
@Test
public void testMultiFieldsSequenceGroup() {
sql(
