diff --git a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java index 92b675c428c..ee779b7d885 100644 --- a/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java +++ b/fluss-common/src/main/java/org/apache/fluss/metadata/Schema.java @@ -37,6 +37,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.EnumSet; import java.util.HashSet; import java.util.List; import java.util.Objects; @@ -65,6 +66,7 @@ public final class Schema implements Serializable { private final List columns; private final @Nullable PrimaryKey primaryKey; private final List autoIncrementColumnNames; + private final List sequenceGroups; private final RowType rowType; /** @@ -77,11 +79,13 @@ private Schema( List columns, @Nullable PrimaryKey primaryKey, int highestFieldId, - List autoIncrementColumnNames) { + List autoIncrementColumnNames, + List sequenceGroups) { this.columns = normalizeColumns(columns, primaryKey, autoIncrementColumnNames, highestFieldId); this.primaryKey = primaryKey; this.autoIncrementColumnNames = autoIncrementColumnNames; + this.sequenceGroups = Collections.unmodifiableList(new ArrayList<>(sequenceGroups)); // pre-create the row type as it is the most frequently used part of the schema this.rowType = new RowType( @@ -94,6 +98,15 @@ private Schema( column.columnId)) .collect(Collectors.toList())); this.highestFieldId = highestFieldId; + + // enforce every schema-level sequence group invariant here so that a built Schema is + // always consistent; merge-engine and log-table rejections still run at table creation + validateSequenceGroups( + this.rowType, + this.primaryKey, + this.autoIncrementColumnNames, + this.columns, + this.sequenceGroups); } public List getColumns() { @@ -142,6 +155,19 @@ public Optional getAggFunction(String columnName) { .flatMap(Column::getAggFunction); } + /** Returns true if at least one column of this schema is protected by a sequence group. */ + public boolean hasSequenceGroup() { + return !sequenceGroups.isEmpty(); + } + + /** + * Gets the sequence groups declared on this schema, empty when it declares none. Each group + * relates an ordered list of sequence columns to the columns they protect. + */ + public List getSequenceGroups() { + return sequenceGroups; + } + /** Returns the primary key indexes, if any, otherwise returns an empty array. */ public int[] getPrimaryKeyIndexes() { final List columns = getColumnNames(); @@ -233,6 +259,8 @@ public String toString() { + primaryKey + ", autoIncrementColumnNames=" + autoIncrementColumnNames + + ", sequenceGroups=" + + sequenceGroups + ", highestFieldId=" + highestFieldId + '}'; @@ -250,12 +278,14 @@ public boolean equals(Object o) { return Objects.equals(columns, schema.columns) && Objects.equals(autoIncrementColumnNames, schema.autoIncrementColumnNames) && Objects.equals(primaryKey, schema.primaryKey) + && Objects.equals(sequenceGroups, schema.sequenceGroups) && highestFieldId == schema.highestFieldId; } @Override public int hashCode() { - return Objects.hash(columns, primaryKey, autoIncrementColumnNames, highestFieldId); + return Objects.hash( + columns, primaryKey, autoIncrementColumnNames, sequenceGroups, highestFieldId); } // -------------------------------------------------------------------------------------------- @@ -275,11 +305,13 @@ public static final class Builder { private final List columns; private @Nullable PrimaryKey primaryKey; private final List autoIncrementColumnNames; + private final List sequenceGroups; private AtomicInteger highestFieldId; private Builder() { columns = new ArrayList<>(); autoIncrementColumnNames = new ArrayList<>(); + sequenceGroups = new ArrayList<>(); highestFieldId = new AtomicInteger(-1); } @@ -298,6 +330,7 @@ public Builder fromSchema(Schema schema) { // Copy the metadata members this.autoIncrementColumnNames.addAll(schema.getAutoIncrementColumnNames()); + this.sequenceGroups.addAll(schema.getSequenceGroups()); schema.getPrimaryKey().ifPresent(pk -> this.primaryKey = pk); return this; @@ -488,6 +521,26 @@ public Builder withComment(@Nullable String comment) { return this; } + /** + * Declares a sequence group, relating an ordered list of sequence columns to the columns + * they protect. A protected column then only takes an incoming value when the sequence + * columns are not older than the stored ones, and every group is arbitrated on its own. + * + *

Passing more than one sequence column declares a composite sequence key, where the + * columns are compared in the given order and the first unequal one decides. + * + * @param sequenceColumns the columns ordering the group, in comparison order + * @param protectedColumns the columns held under that order + */ + public Builder sequenceGroup(List sequenceColumns, List protectedColumns) { + checkNotNull(sequenceColumns, "Sequence columns must not be null."); + checkNotNull(protectedColumns, "Protected columns must not be null."); + checkArgument(!sequenceColumns.isEmpty(), "Sequence columns must not be empty."); + checkArgument(!protectedColumns.isEmpty(), "Protected columns must not be empty."); + sequenceGroups.add(new SequenceGroup(sequenceColumns, protectedColumns)); + return this; + } + /** * Declares a primary key constraint for a set of given columns. Primary key uniquely * identify a row in a table. Neither of columns in a primary can be nullable. Adding a @@ -567,7 +620,12 @@ public Optional getColumn(String columnName) { /** Returns an instance of an {@link Schema}. */ public Schema build() { - return new Schema(columns, primaryKey, highestFieldId.get(), autoIncrementColumnNames); + return new Schema( + columns, + primaryKey, + highestFieldId.get(), + autoIncrementColumnNames, + sequenceGroups); } } @@ -575,6 +633,65 @@ public Schema build() { // Helper classes for representing the schema // -------------------------------------------------------------------------------------------- + /** + * A sequence group, relating an ordered list of sequence columns to the columns they protect. + * + *

The order of the sequence columns is semantically significant because it defines the + * composite comparison key. The protected columns have set semantics, although the declaration + * order is preserved for deterministic serialization. + * + * @since 0.7 + */ + @PublicStable + public static final class SequenceGroup implements Serializable { + private static final long serialVersionUID = 1L; + private final List sequenceColumns; + private final List protectedColumns; + + public SequenceGroup(List sequenceColumns, List protectedColumns) { + this.sequenceColumns = Collections.unmodifiableList(new ArrayList<>(sequenceColumns)); + this.protectedColumns = Collections.unmodifiableList(new ArrayList<>(protectedColumns)); + } + + /** The columns defining the comparison key, in comparison order. */ + public List getSequenceColumns() { + return sequenceColumns; + } + + /** The columns held under this order. */ + public List getProtectedColumns() { + return protectedColumns; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + SequenceGroup that = (SequenceGroup) o; + return Objects.equals(sequenceColumns, that.sequenceColumns) + && Objects.equals(protectedColumns, that.protectedColumns); + } + + @Override + public int hashCode() { + return Objects.hash(sequenceColumns, protectedColumns); + } + + @Override + public String toString() { + return "SequenceGroup{" + + "sequenceColumns=" + + sequenceColumns + + ", protectedColumns=" + + protectedColumns + + '}'; + } + } + /** * column in a schema. * @@ -835,6 +952,175 @@ private static Set duplicate(List names) { .collect(Collectors.toSet()); } + /** + * Rejects a sequence group configuration that would leave the schema in an inconsistent state. + * Everything checked here can be decided from the schema alone; merge-engine and log-table + * rejections are left to {@code TableDescriptorValidation} at table creation. + */ + private static void validateSequenceGroups( + RowType rowType, + @Nullable PrimaryKey primaryKey, + List autoIncrementColumnNames, + List columns, + List sequenceGroups) { + if (sequenceGroups.isEmpty()) { + return; + } + + for (SequenceGroup group : sequenceGroups) { + if (group.getSequenceColumns().isEmpty()) { + throw new IllegalArgumentException( + "The sequence columns of a sequence group must not be empty."); + } + if (group.getProtectedColumns().isEmpty()) { + throw new IllegalArgumentException( + "The protected columns of a sequence group must not be empty."); + } + } + + List primaryKeyNames = + primaryKey == null ? Collections.emptyList() : primaryKey.getColumnNames(); + EnumSet supportedTypes = + EnumSet.of( + DataTypeRoot.INTEGER, + DataTypeRoot.BIGINT, + DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE, + DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE); + + // collect the columns declared as sequence and as protected across every group, so that we + // can reject a column belonging to more than one group or crossing the two roles + Set allSequenceColumns = new HashSet<>(); + Set allProtectedColumns = new HashSet<>(); + + for (SequenceGroup group : sequenceGroups) { + List sequenceColumns = group.getSequenceColumns(); + List protectedColumns = group.getProtectedColumns(); + + rejectDuplicateWithinGroup(sequenceColumns, "sequence"); + rejectDuplicateWithinGroup(protectedColumns, "protected"); + + for (String sequenceColumn : sequenceColumns) { + int columnIndex = rowType.getFieldIndex(sequenceColumn); + if (columnIndex < 0) { + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' doesn't exist in schema.", + sequenceColumn)); + } + if (primaryKeyNames.contains(sequenceColumn)) { + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' must not be a primary key column.", + sequenceColumn)); + } + // the group it orders decides when it advances, so an aggregate function on it + // would let a stale row move the sequence backwards + if (aggFunctionOf(columns, sequenceColumn).isPresent()) { + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' orders a sequence group, " + + "so it must not have an aggregate function.", + sequenceColumn)); + } + DataType columnType = rowType.getTypeAt(columnIndex); + if (!supportedTypes.contains(columnType.getTypeRoot())) { + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' must be one type of " + + "[INT, BIGINT, TIMESTAMP, TIMESTAMP_LTZ], but got %s.", + sequenceColumn, columnType)); + } + // a sequence column names the order of one group only. This also rejects two + // groups sharing all their sequence columns, which are the same group twice. + if (!allSequenceColumns.add(sequenceColumn)) { + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' must not be shared by more than one sequence group.", + sequenceColumn)); + } + } + + for (String protectedColumn : protectedColumns) { + int columnIndex = rowType.getFieldIndex(protectedColumn); + if (columnIndex < 0) { + throw new IllegalArgumentException( + String.format( + "The protected column '%s' doesn't exist in schema.", + protectedColumn)); + } + // a primary key column holds the same value in both rows being merged, so it can + // neither be held back by a group nor order one + if (primaryKeyNames.contains(protectedColumn)) { + throw new IllegalArgumentException( + String.format( + "The primary key column '%s' must not be put in a sequence group.", + protectedColumn)); + } + if (!allProtectedColumns.add(protectedColumn)) { + throw new IllegalArgumentException( + String.format( + "The column '%s' must not be protected by more than one sequence group.", + protectedColumn)); + } + } + } + + // a sequence column reports the order of its own group, so it cannot also be held back by + // another one + for (String sequenceColumn : allSequenceColumns) { + if (allProtectedColumns.contains(sequenceColumn)) { + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' orders a sequence group, " + + "so it must not be put into another one.", + sequenceColumn)); + } + } + + // a client may not write an auto increment column, so a group containing one could never + // be updated. + for (String groupField : allSequenceColumns) { + if (autoIncrementColumnNames.contains(groupField)) { + throw new IllegalArgumentException( + String.format( + "The auto increment column '%s' must not order a sequence group.", + groupField)); + } + } + for (String groupField : allProtectedColumns) { + if (autoIncrementColumnNames.contains(groupField)) { + throw new IllegalArgumentException( + String.format( + "The auto increment column '%s' must not be put in a sequence group.", + groupField)); + } + } + } + + private static Optional aggFunctionOf(List columns, String name) { + return columns.stream() + .filter(column -> column.getName().equals(name)) + .findFirst() + .flatMap(Column::getAggFunction); + } + + /** + * Rejects a column named more than once in the same list, since a repeated name always signals + * a typo: naming a column twice as a sequence column would degenerate the comparison key, and + * naming it twice as a protected column adds nothing. + */ + private static void rejectDuplicateWithinGroup(List names, String role) { + Set seen = new HashSet<>(); + for (String name : names) { + if (!seen.add(name)) { + throw new IllegalArgumentException( + String.format( + "The %s column '%s' is declared more than once in the same sequence group.", + role, name)); + } + } + } + public static RowType getKeyRowType(Schema schema, int[] keyIndexes) { List keyRowFields = new ArrayList<>(keyIndexes.length); List rowFields = schema.getRowType().getFields(); diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/SchemaJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/SchemaJsonSerde.java index 6fd1ddb75ce..ee5c164b39e 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/json/SchemaJsonSerde.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/SchemaJsonSerde.java @@ -37,6 +37,9 @@ public class SchemaJsonSerde implements JsonSerializer, JsonDeserializer private static final String COLUMNS_NAME = "columns"; private static final String PRIMARY_KEY_NAME = "primary_key"; private static final String AUTO_INCREMENT_COLUMN_NAME = "auto_increment_column"; + private static final String SEQUENCE_GROUPS_NAME = "sequence_groups"; + private static final String SEQUENCE_COLUMNS_NAME = "sequence_columns"; + private static final String PROTECTED_COLUMNS_NAME = "protected_columns"; private static final String VERSION_KEY = "version"; private static final String HIGHEST_FIELD_ID = "highest_field_id"; private static final int VERSION = 1; @@ -72,6 +75,26 @@ public void serialize(Schema schema, JsonGenerator generator) throws IOException generator.writeEndArray(); } + List sequenceGroups = schema.getSequenceGroups(); + if (!sequenceGroups.isEmpty()) { + generator.writeArrayFieldStart(SEQUENCE_GROUPS_NAME); + for (Schema.SequenceGroup group : sequenceGroups) { + generator.writeStartObject(); + generator.writeArrayFieldStart(SEQUENCE_COLUMNS_NAME); + for (String column : group.getSequenceColumns()) { + generator.writeString(column); + } + generator.writeEndArray(); + generator.writeArrayFieldStart(PROTECTED_COLUMNS_NAME); + for (String column : group.getProtectedColumns()) { + generator.writeString(column); + } + generator.writeEndArray(); + generator.writeEndObject(); + } + generator.writeEndArray(); + } + generator.writeNumberField(HIGHEST_FIELD_ID, schema.getHighestFieldId()); generator.writeEndObject(); @@ -103,10 +126,26 @@ public Schema deserialize(JsonNode node) { } } + if (node.has(SEQUENCE_GROUPS_NAME)) { + for (JsonNode groupJson : node.get(SEQUENCE_GROUPS_NAME)) { + List sequenceColumns = readStringArray(groupJson, SEQUENCE_COLUMNS_NAME); + List protectedColumns = readStringArray(groupJson, PROTECTED_COLUMNS_NAME); + builder.sequenceGroup(sequenceColumns, protectedColumns); + } + } + if (node.has(HIGHEST_FIELD_ID)) { builder.highestFieldId(node.get(HIGHEST_FIELD_ID).asInt()); } return builder.build(); } + + private static List readStringArray(JsonNode parent, String field) { + List values = new ArrayList<>(); + for (JsonNode element : parent.get(field)) { + values.add(element.asText()); + } + return values; + } } diff --git a/fluss-common/src/test/java/org/apache/fluss/metadata/SchemaSequenceGroupTest.java b/fluss-common/src/test/java/org/apache/fluss/metadata/SchemaSequenceGroupTest.java new file mode 100644 index 00000000000..d07f2af02ba --- /dev/null +++ b/fluss-common/src/test/java/org/apache/fluss/metadata/SchemaSequenceGroupTest.java @@ -0,0 +1,240 @@ +/* + * 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.fluss.metadata; + +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.stream.Stream; + +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Schema-level rejection tests for sequence groups. These checks are decided from the schema alone, + * so they run inside {@link Schema.Builder#build()}. Table-level rejections (merge engine, log + * table) still live in {@code SequenceGroupValidationTest}. + */ +class SchemaSequenceGroupTest { + + private static Schema.Builder pkSchema() { + return Schema.newBuilder().column("k", DataTypes.INT()).column("a", DataTypes.STRING()); + } + + private static Stream supportedSequenceTypes() { + return Stream.of( + DataTypes.INT(), + DataTypes.BIGINT(), + DataTypes.TIMESTAMP(), + DataTypes.TIMESTAMP_LTZ()); + } + + @ParameterizedTest + @MethodSource("supportedSequenceTypes") + void testSupportedSequenceColumnTypeIsAccepted(DataType sequenceType) { + assertThatCode( + () -> + pkSchema() + .column("g", sequenceType) + .sequenceGroup(singletonList("g"), singletonList("a")) + .primaryKey("k") + .build()) + .doesNotThrowAnyException(); + } + + @Test + void testEveryColumnOfACompositeSequenceKeyIsChecked() { + assertThatThrownBy( + () -> + pkSchema() + .column("g1", DataTypes.INT()) + // only the trailing column has an unsupported type + .column("g2", DataTypes.STRING()) + .sequenceGroup(asList("g1", "g2"), singletonList("a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The sequence column 'g2' must be one type of " + + "[INT, BIGINT, TIMESTAMP, TIMESTAMP_LTZ], but got STRING"); + } + + @Test + void testUnknownSequenceColumnIsRejected() { + assertThatThrownBy( + () -> + pkSchema() + .sequenceGroup(singletonList("missing"), singletonList("a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("The sequence column 'missing' doesn't exist in schema."); + } + + @Test + void testUnknownProtectedColumnIsRejected() { + assertThatThrownBy( + () -> + pkSchema() + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), singletonList("missing")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("The protected column 'missing' doesn't exist in schema."); + } + + @Test + void testSequenceColumnWithAggregateFunctionIsRejected() { + assertThatThrownBy( + () -> + pkSchema() + .column( + "g", + DataTypes.INT(), + AggFunctions.of(AggFunctionType.SUM)) + .sequenceGroup(singletonList("g"), singletonList("a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The sequence column 'g' orders a sequence group, " + + "so it must not have an aggregate function."); + } + + @Test + void testPrimaryKeyColumnInSequenceGroupIsRejected() { + // a primary key holds the same value in both rows being merged, so a group can neither + // arbitrate it nor be ordered by it + assertThatThrownBy( + () -> + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), singletonList("k")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The primary key column 'k' must not be put in a sequence group."); + } + + @Test + void testPrimaryKeyAsSequenceColumnIsRejected() { + assertThatThrownBy( + () -> + pkSchema() + .sequenceGroup(singletonList("k"), singletonList("a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("The sequence column 'k' must not be a primary key column."); + } + + @Test + void testSequenceColumnProtectedByAnotherGroupIsRejected() { + // a sequence column reports the order of its own group, so following another one would + // leave it out of step with the columns it orders + assertThatThrownBy( + () -> + pkSchema() + .column("pay_time", DataTypes.TIMESTAMP()) + .column("ship_time", DataTypes.TIMESTAMP()) + .sequenceGroup( + singletonList("pay_time"), singletonList("a")) + .sequenceGroup( + singletonList("ship_time"), + singletonList("pay_time")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The sequence column 'pay_time' orders a sequence group, " + + "so it must not be put into another one."); + } + + @Test + void testRepeatedSequenceColumnWithinGroupIsRejected() { + // naming a column twice as a sequence column degenerates the comparison key, so the intent + // is always a typo + assertThatThrownBy( + () -> + pkSchema() + .column("g", DataTypes.INT()) + .sequenceGroup(asList("g", "g"), singletonList("a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The sequence column 'g' is declared more than once in the same sequence group."); + } + + @Test + void testRepeatedProtectedColumnWithinGroupIsRejected() { + assertThatThrownBy( + () -> + pkSchema() + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), asList("a", "a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The protected column 'a' is declared more than once in the same sequence group."); + } + + @Test + void testSequenceColumnSharedByTwoGroupsIsRejected() { + // a sequence column shared by two groups would define two different orders; this also + // rejects two groups with identical sequence columns, which are the same group twice. + assertThatThrownBy( + () -> + pkSchema() + .column("b", DataTypes.STRING()) + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), singletonList("a")) + .sequenceGroup(singletonList("g"), singletonList("b")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The sequence column 'g' must not be shared by more than one sequence group."); + } + + @Test + void testColumnProtectedByTwoGroupsIsRejected() { + assertThatThrownBy( + () -> + pkSchema() + .column("g1", DataTypes.INT()) + .column("g2", DataTypes.INT()) + .sequenceGroup(singletonList("g1"), singletonList("a")) + .sequenceGroup(singletonList("g2"), singletonList("a")) + .primaryKey("k") + .build()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "The column 'a' must not be protected by more than one sequence group."); + } +} diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java index c200db85735..038176e32a9 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java @@ -62,6 +62,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -95,6 +96,9 @@ /** Utils for conversion between Flink and Fluss. */ public class FlinkConversions { + private static final String SEQUENCE_GROUP_PREFIX = "fields."; + private static final String SEQUENCE_GROUP_SUFFIX = ".sequence-group"; + private FlinkConversions() {} /** Convert Fluss's type to Flink's type. */ @@ -144,6 +148,16 @@ public static CatalogBaseTable toFlinkTable(TableInfo tableInfo) { column.getName(), column.getAggFunction().get(), newOptions); } } + + // rebuild the fields..sequence-group options from the schema, so that a + // SHOW CREATE TABLE reflects the declarations kept on Schema.sequenceGroups + for (Schema.SequenceGroup group : schema.getSequenceGroups()) { + String key = + SEQUENCE_GROUP_PREFIX + + String.join(",", group.getSequenceColumns()) + + SEQUENCE_GROUP_SUFFIX; + newOptions.put(key, String.join(",", group.getProtectedColumns())); + } List physicalColumns = schema.getColumnNames(); int columnCount = physicalColumns.size() @@ -225,6 +239,13 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable catalogBa addColumnToSchema( schemBuilder, column, flinkTableConf, isAggregationEngine)); + // Sequence groups are a cross-column relation, so they are added on the schema itself + // rather than on the individual protected columns + parseSequenceGroups(flinkTableConf) + .forEach( + (sequenceColumns, protectedColumns) -> + schemBuilder.sequenceGroup(sequenceColumns, protectedColumns)); + // Configure auto-increment columns based on the 'auto-increment.fields' option. if (flinkTableConf.containsKey(AUTO_INCREMENT_FIELDS.key())) { for (String autoIncrementColumn : @@ -744,7 +765,102 @@ private static boolean isAggregationMergeEngine(Configuration tableConf) { } /** - * Add a column to the schema builder with optional aggregation function. + * Parses the sequence groups declared in the table options. The key lists the sequence columns + * ordering the group and the value lists the columns it protects: + * + * 

+     * 'fields.g1.sequence-group' = 'a,b'
+     * 'fields.g1,g2.sequence-group' = 'c'
+     *
+ * + *

Returns the groups in a map from ordered sequence columns to protected columns, keeping + * the shape declared in the DDL so that the schema can hold it directly. + */ + private static Map, List> parseSequenceGroups(Configuration tableConf) { + // LinkedHashMap keeps the group order stable across serializations + Map, List> groups = new LinkedHashMap<>(); + Map> sequenceColumnsOfProtected = new HashMap<>(); + for (String key : tableConf.keySet()) { + if (!key.startsWith(SEQUENCE_GROUP_PREFIX) || !key.endsWith(SEQUENCE_GROUP_SUFFIX)) { + continue; + } + List sequenceColumns = + splitColumns( + key.substring( + SEQUENCE_GROUP_PREFIX.length(), + key.length() - SEQUENCE_GROUP_SUFFIX.length()), + key, + "sequence columns"); + // the key comes from the option keys, so a value is always present + List protectedColumns = + splitColumns(tableConf.getString(key, ""), key, "protected columns"); + + for (String protectedColumn : protectedColumns) { + if (sequenceColumns.contains(protectedColumn)) { + throw new IllegalArgumentException( + String.format( + "Invalid option '%s': column '%s' must not be protected by itself.", + key, protectedColumn)); + } + List previous = + sequenceColumnsOfProtected.put(protectedColumn, sequenceColumns); + if (previous != null) { + throw new IllegalArgumentException( + String.format( + "Column '%s' is declared repeatedly by sequence groups %s and %s.", + protectedColumn, previous, sequenceColumns)); + } + } + groups.merge( + sequenceColumns, + protectedColumns, + (a, b) -> { + List merged = new ArrayList<>(a); + for (String c : b) { + if (!merged.contains(c)) { + merged.add(c); + } + } + return merged; + }); + } + return groups; + } + + /** + * Splits a comma separated list of column names, rejecting an empty list as well as an empty or + * repeated name. A repeated name is always a typo, since naming a column twice adds nothing. + * + * @param description names the parsed list in the rejection message + */ + private static List splitColumns(String value, String key, String description) { + if (value.trim().isEmpty()) { + throw new IllegalArgumentException( + String.format( + "Invalid option '%s': the %s must not be empty.", key, description)); + } + List columns = new ArrayList<>(); + for (String column : value.split(",")) { + String trimmed = column.trim(); + if (trimmed.isEmpty()) { + throw new IllegalArgumentException( + String.format( + "Invalid option '%s': the %s must not be empty.", + key, description)); + } + if (columns.contains(trimmed)) { + throw new IllegalArgumentException( + String.format( + "Invalid option '%s': the %s name '%s' is declared more than once.", + key, description, trimmed)); + } + columns.add(trimmed); + } + return columns; + } + + /** + * Add a column to the schema builder with optional aggregation function and sequence columns. * * @param schemaBuilder the schema builder * @param column the Flink column @@ -784,6 +900,15 @@ private static Map extractCustomProperties( // properties. customProperties.remove(BUCKET_KEY.key()); customProperties.remove(BUCKET_NUMBER.key()); + // Sequence group options are consumed into Schema.sequenceGroups, so they must not be + // retained as custom properties as well. Otherwise the table would carry two + // representations that could drift apart across ALTER TABLE SET. + customProperties + .keySet() + .removeIf( + key -> + key.startsWith(SEQUENCE_GROUP_PREFIX) + && key.endsWith(SEQUENCE_GROUP_SUFFIX)); return customProperties; } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java index 09a59de57ff..7a4b51ad583 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java @@ -2002,4 +2002,159 @@ void testWalModeWithAutoIncrement() throws Exception { * version. The Flink 2.3-specific subclass overrides it to actually flip the option off. */ protected void disableSinkRequireOnConflict() {} + + @Test + void testSequenceGroupArbitratesEachGroupOnItsOwn() throws Exception { + // the groups are declared on the table, so they have to survive being persisted to and read + // back from the server before any of this can arbitrate a write + tEnv.executeSql( + "create table seq_group (" + + " k int not null primary key not enforced," + + " pay_status string, pay_time bigint," + + " ship_status string, ship_time bigint" + + ") with ('fields.pay_time.sequence-group' = 'pay_status'," + + "'fields.ship_time.sequence-group' = 'ship_status')"); + + tEnv.executeSql("insert into seq_group values (1, 'paid', 100, 'shipped', 100)").await(); + + CloseableIterator rowIter = tEnv.executeSql("select * from seq_group").collect(); + assertResultsIgnoreOrder( + rowIter, Collections.singletonList("+I[1, paid, 100, shipped, 100]"), false); + + // the pay group moves forward while the ship group falls behind, so only the pay columns + // take the incoming values + tEnv.executeSql("insert into seq_group values (1, 'refunded', 200, 'lost', 99)").await(); + assertResultsIgnoreOrder( + rowIter, + Arrays.asList( + "-U[1, paid, 100, shipped, 100]", "+U[1, refunded, 200, shipped, 100]"), + false); + + // the ship group catches up on its own, leaving the pay columns untouched + tEnv.executeSql("insert into seq_group values (1, 'stale', 2, 'delivered', 300)").await(); + assertResultsIgnoreOrder( + rowIter, + Arrays.asList( + "-U[1, refunded, 200, shipped, 100]", + "+U[1, refunded, 200, delivered, 300]"), + true); + } + + @Test + void testSequenceGroupSurvivesAddColumn() throws Exception { + tEnv.executeSql( + "create table seq_group_evolving (" + + " k int not null primary key not enforced, v string, ts bigint" + + ") with ('fields.ts.sequence-group' = 'v')"); + + tEnv.executeSql("insert into seq_group_evolving values (1, 'first', 100)").await(); + tEnv.executeSql("alter table seq_group_evolving add (extra string)"); + + CloseableIterator rowIter = + tEnv.executeSql("select * from seq_group_evolving").collect(); + assertResultsIgnoreOrder( + rowIter, Collections.singletonList("+I[1, first, 100, null]"), false); + + // the group keeps arbitrating across the schema change, and the row stored under the older + // schema is read back with the added column as null + tEnv.executeSql("insert into seq_group_evolving values (1, 'newer', 101, 'x')").await(); + assertResultsIgnoreOrder( + rowIter, Arrays.asList("-U[1, first, 100, null]", "+U[1, newer, 101, x]"), true); + } + + @Test + void testUnsupportedSequenceGroupIsRejectedByTheServer() { + // the client parses the declaration while only the server can judge it, so the rejection + // has to travel back across that boundary + assertThatThrownBy( + () -> + tEnv.executeSql( + "create table seq_group_bad_type (" + + " k int not null primary key not enforced," + + " v string, ts string)" + + " with ('fields.ts.sequence-group' = 'v')")) + .rootCause() + .hasMessageContaining("The sequence column 'ts' must be one type of"); + } + + @Test + void testSequenceGroupOnAggregationMergeEngine() throws Exception { + // with an aggregate function a sequence group orders the records rather than filtering + // them: a stale record still contributes to the sum, it only must not move the sequence + tEnv.executeSql( + "create table agg_seq_group (" + + " k int not null primary key not enforced," + + " total bigint," + + " ts int" + + ") with ('table.merge-engine' = 'aggregation'," + + "'fields.total.agg' = 'sum'," + + "'fields.ts.sequence-group' = 'total')"); + + tEnv.executeSql("insert into agg_seq_group values (1, 30, 100)").await(); + + CloseableIterator rowIter = tEnv.executeSql("select * from agg_seq_group").collect(); + assertResultsIgnoreOrder(rowIter, Collections.singletonList("+I[1, 30, 100]"), false); + + // the sequence moves forward, so the total accumulates and the sequence follows + tEnv.executeSql("insert into agg_seq_group values (1, 20, 200)").await(); + assertResultsIgnoreOrder(rowIter, Arrays.asList("-U[1, 30, 100]", "+U[1, 50, 200]"), false); + + // an older record still accumulates, but leaves the stored sequence at 200 + tEnv.executeSql("insert into agg_seq_group values (1, 10, 50)").await(); + assertResultsIgnoreOrder(rowIter, Arrays.asList("-U[1, 50, 200]", "+U[1, 60, 200]"), false); + + // a record without any sequence carries no order information, so it contributes nothing: + // the write changes nothing and produces no changelog event at all + tEnv.executeSql("insert into agg_seq_group values (1, 5, cast(null as int))").await(); + assertResultsIgnoreOrder(rowIter, Collections.emptyList(), true); + } + + @Test + void testSequenceGroupsAreArbitratedIndependentlyOnAggregationMergeEngine() throws Exception { + tEnv.executeSql( + "create table agg_two_groups (" + + " k int not null primary key not enforced," + + " paid bigint, pay_ts int," + + " shipped bigint, ship_ts int" + + ") with ('table.merge-engine' = 'aggregation'," + + "'fields.paid.agg' = 'sum'," + + "'fields.shipped.agg' = 'sum'," + + "'fields.pay_ts.sequence-group' = 'paid'," + + "'fields.ship_ts.sequence-group' = 'shipped')"); + + tEnv.executeSql("insert into agg_two_groups values (1, 30, 100, 30, 100)").await(); + + CloseableIterator rowIter = tEnv.executeSql("select * from agg_two_groups").collect(); + assertResultsIgnoreOrder( + rowIter, Collections.singletonList("+I[1, 30, 100, 30, 100]"), false); + + // the pay group moves forward while the ship group carries no sequence at all, so only the + // pay total accumulates + tEnv.executeSql( + "insert into agg_two_groups values " + + "(1, 20, 200, 20, cast(null as int))") + .await(); + assertResultsIgnoreOrder( + rowIter, Arrays.asList("-U[1, 30, 100, 30, 100]", "+U[1, 50, 200, 30, 100]"), true); + } + + @Test + void testSequenceColumnWithAggregateFunctionIsRejectedByTheServer() { + // the group it orders decides when it advances, so aggregating the sequence column itself + // would let a stale record move the sequence backwards + assertThatThrownBy( + () -> + tEnv.executeSql( + "create table agg_seq_bad (" + + " k int not null primary key not enforced," + + " total bigint, ts int)" + + " with ('table.merge-engine' = 'aggregation'," + + "'fields.total.agg' = 'sum'," + + "'fields.ts.agg' = 'sum'," + + "'fields.ts.sequence-group' = 'total')")) + .rootCause() + .hasMessageContaining( + "The sequence column 'ts' orders a sequence group, " + + "so it must not have an aggregate function."); + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java index b70a4d88e4b..45daaff4baf 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java @@ -203,7 +203,10 @@ void testBoundedPkTableEmitsKvBatchSplits() throws Throwable { @Test void testBoundedPkTableEmitsSnapshotSplitsByDefault() throws Throwable { - createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR); + long tableId = createTable(DEFAULT_TABLE_PATH, DEFAULT_PK_TABLE_DESCRIPTOR); + // creating a table returns before its buckets have elected a leader, and this test lists + // offsets right away, which needs the leader to be in the metadata cache + FLUSS_CLUSTER_EXTENSION.waitUntilTableReady(tableId); int numSubtasks = DEFAULT_BUCKET_NUM; try (MockSplitEnumeratorContext context = new MockSplitEnumeratorContext<>(numSubtasks)) { diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java index 71a81d3a079..f204aa13708 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java @@ -228,7 +228,7 @@ void testTableConversion() { String expectFlussTableString = "TableDescriptor{schema=Schema{columns=[order_id STRING NOT NULL, item ROW<`item_id` STRING, `item_price` STRING, `item_details` ROW<`category` STRING, `specifications` STRING>>, orig_ts TIMESTAMP(6)], " + "primaryKey=CONSTRAINT PK_order_id PRIMARY KEY (order_id), " - + "autoIncrementColumnNames=[], highestFieldId=7}, comment='test comment', partitionKeys=[], " + + "autoIncrementColumnNames=[], sequenceGroups=[], highestFieldId=7}, comment='test comment', partitionKeys=[], " + "tableDistribution={bucketKeys=[order_id] bucketCount=null}, " + "properties={}, " + "customProperties={" @@ -359,6 +359,166 @@ void testTableConversionForCustomProperties() { assertThat(flussTable.getCustomProperties()).containsExactlyEntriesOf(customProperties); } + /** + * Converts a primary key table declaring the given options, whose columns are {@code k}, {@code + * a}, {@code b}, {@code g1} and {@code g2}. + */ + private static org.apache.fluss.metadata.Schema convertWithOptions( + Map options) { + ResolvedSchema resolvedSchema = + new ResolvedSchema( + Arrays.asList( + Column.physical( + "k", + org.apache.flink.table.api.DataTypes.BIGINT().notNull()), + Column.physical("a", org.apache.flink.table.api.DataTypes.STRING()), + Column.physical("b", org.apache.flink.table.api.DataTypes.STRING()), + Column.physical( + "g1", org.apache.flink.table.api.DataTypes.BIGINT()), + Column.physical( + "g2", org.apache.flink.table.api.DataTypes.BIGINT())), + Collections.emptyList(), + null); + CatalogTable flinkTable = + CatalogTable.of( + Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(), + null, + Collections.emptyList(), + options); + return FlinkConversions.toFlussTable(new ResolvedCatalogTable(flinkTable, resolvedSchema)) + .getSchema(); + } + + private static Map sequenceGroup(String key, String value) { + Map options = new HashMap<>(); + options.put(key, value); + return options; + } + + private static List sequenceColumnsOf( + org.apache.fluss.metadata.Schema schema, String protectedColumn) { + return schema.getSequenceGroups().stream() + .filter(group -> group.getProtectedColumns().contains(protectedColumn)) + .findFirst() + .map(org.apache.fluss.metadata.Schema.SequenceGroup::getSequenceColumns) + .orElse(null); + } + + private static List protectedColumnsOf( + org.apache.fluss.metadata.Schema schema, List sequenceColumns) { + return schema.getSequenceGroups().stream() + .filter(group -> group.getSequenceColumns().equals(sequenceColumns)) + .findFirst() + .map(org.apache.fluss.metadata.Schema.SequenceGroup::getProtectedColumns) + .orElse(null); + } + + private static void assertSequenceGroupRejected(String key, String value, String message) { + assertThatThrownBy(() -> convertWithOptions(sequenceGroup(key, value))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(message); + } + + @Test + void testSequenceGroupIsAttachedToTheSchema() { + org.apache.fluss.metadata.Schema schema = + convertWithOptions(sequenceGroup("fields.g1.sequence-group", "a, b")); + + // the declaration lives on Schema.sequenceGroups as a cross-column relation, so it is not + // duplicated on every protected column, and the names are trimmed on the way + assertThat(schema.hasSequenceGroup()).isTrue(); + assertThat(schema.getSequenceGroups()).hasSize(1); + org.apache.fluss.metadata.Schema.SequenceGroup group = schema.getSequenceGroups().get(0); + assertThat(group.getSequenceColumns()).containsExactly("g1"); + assertThat(group.getProtectedColumns()).containsExactly("a", "b"); + + // the helper reads the same information from a protected column's point of view + assertThat(sequenceColumnsOf(schema, "a")).containsExactly("g1"); + assertThat(sequenceColumnsOf(schema, "b")).containsExactly("g1"); + assertThat(sequenceColumnsOf(schema, "k")).isNull(); + assertThat(sequenceColumnsOf(schema, "g1")).isNull(); + } + + @Test + void testCompositeSequenceGroupKeepsItsDeclaredOrder() { + // the order the sequence columns are named in is the order they are compared in + org.apache.fluss.metadata.Schema schema = + convertWithOptions(sequenceGroup("fields. g2 , g1 .sequence-group", "a")); + + assertThat(sequenceColumnsOf(schema, "a")).containsExactly("g2", "g1"); + assertThat(protectedColumnsOf(schema, java.util.Arrays.asList("g2", "g1"))) + .containsExactly("a"); + } + + @Test + void testColumnDeclaredByTwoSequenceGroupsIsRejected() { + Map options = sequenceGroup("fields.g1.sequence-group", "a"); + options.put("fields.g2.sequence-group", "a"); + + assertThatThrownBy(() -> convertWithOptions(options)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("is declared repeatedly by sequence groups"); + } + + @Test + void testColumnProtectedByItselfIsRejected() { + assertSequenceGroupRejected( + "fields.g1.sequence-group", "a,g1", "column 'g1' must not be protected by itself"); + } + + @Test + void testSequenceGroupSurvivesTheRoundTrip() { + // starting from the DDL, the option becomes a schema group and the schema group is + // then rebuilt back into an equivalent option. The option must not survive on + // customProperties, otherwise the two representations would be free to drift apart. + Map declared = new HashMap<>(); + declared.put("fields.g1,g2.sequence-group", "a,b"); + + ResolvedSchema resolvedSchema = + new ResolvedSchema( + Arrays.asList( + Column.physical( + "k", + org.apache.flink.table.api.DataTypes.BIGINT().notNull()), + Column.physical("a", org.apache.flink.table.api.DataTypes.STRING()), + Column.physical("b", org.apache.flink.table.api.DataTypes.STRING()), + Column.physical( + "g1", org.apache.flink.table.api.DataTypes.BIGINT()), + Column.physical( + "g2", org.apache.flink.table.api.DataTypes.BIGINT())), + Collections.emptyList(), + null); + CatalogTable flinkTable = + CatalogTable.of( + Schema.newBuilder().fromResolvedSchema(resolvedSchema).build(), + null, + Collections.emptyList(), + declared); + + TableDescriptor flussTable = + FlinkConversions.toFlussTable(new ResolvedCatalogTable(flinkTable, resolvedSchema)); + + // the option is fully consumed into the schema and does not linger on customProperties + assertThat(flussTable.getCustomProperties()) + .doesNotContainKey("fields.g1,g2.sequence-group"); + assertThat(flussTable.getSchema().getSequenceGroups()).hasSize(1); + + TableInfo tableInfo = + TableInfo.of( + TablePath.of("db", "t"), + 1L, + 1, + flussTable.withBucketCount(1), + DEFAULT_REMOTE_DATA_DIR, + 0L, + 0L); + CatalogTable rebuilt = (CatalogTable) FlinkConversions.toFlinkTable(tableInfo); + + // the option is rebuilt back from the schema group, and the round trip therefore preserves + // the DDL a user would see through SHOW CREATE TABLE + assertThat(rebuilt.getOptions()).containsEntry("fields.g1,g2.sequence-group", "a,b"); + } + @Test void testOptionConversions() { ConfigOption flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_KV_FORMAT); @@ -442,7 +602,7 @@ void testFlinkMaterializedTableConversions() { String expectFlussTableString = "TableDescriptor{schema=Schema{columns=[order_id STRING NOT NULL, orig_ts TIMESTAMP(6)], " + "primaryKey=CONSTRAINT PK_order_id PRIMARY KEY (order_id), " - + "autoIncrementColumnNames=[], highestFieldId=1}, comment='test comment', partitionKeys=[], " + + "autoIncrementColumnNames=[], sequenceGroups=[], highestFieldId=1}, comment='test comment', partitionKeys=[], " + "tableDistribution={bucketKeys=[order_id] bucketCount=null}, " + "properties={}, " + "customProperties={materialized-table.definition-query=select order_id, orig_ts from t, " diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java index 76fc94e5c25..edc7c3452ed 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/KvWriteProcessor.java @@ -127,7 +127,7 @@ public KvWriteProcessor( this.rowMerger = rowMerger; // Pre-create DefaultRowMerger for OVERWRITE mode to avoid creating new instances // on every putAsLeader call. Used for undo recovery scenarios. - this.overwriteRowMerger = new DefaultRowMerger(kvFormat, DeleteBehavior.ALLOW); + this.overwriteRowMerger = DefaultRowMerger.forBlindOverwrite(kvFormat); this.arrowCompressionInfo = arrowCompressionInfo; this.schemaGetter = schemaGetter; this.changelogImage = changelogImage; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/TargetColumns.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/TargetColumns.java index 080cf7193fa..bbe07fecd54 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/TargetColumns.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/TargetColumns.java @@ -17,9 +17,13 @@ package org.apache.fluss.server.kv; +import org.apache.fluss.exception.InvalidTargetColumnException; import org.apache.fluss.metadata.Schema; import java.util.BitSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; import static org.apache.fluss.utils.Preconditions.checkNotNull; @@ -55,4 +59,58 @@ public static boolean specifiesAllSchemaFieldIndexes(Schema schema, int[] target } return covered.nextClearBit(0) >= fieldCount; } + + /** + * Rejects a partial write that targets only part of a sequence group. + * + *

A group arbitrates its sequence columns and the columns they protect as one unit, so all + * of them have to move to the incoming row together. Targeting only part of a group would leave + * the stored row with values from two different sequences: either a protected column keeping an + * older value while the group sequence advances, or a protected column taking the incoming + * value on the strength of a sequence that is never stored. + * + * @param schema the schema being written + * @param targetColumns the row field indexes the write targets + * @throws InvalidTargetColumnException if a group is neither fully targeted nor fully left out + */ + public static void checkSequenceGroupsAreFullyTargeted(Schema schema, int[] targetColumns) { + checkNotNull(schema, "schema"); + checkNotNull(targetColumns, "targetColumns"); + List groups = schema.getSequenceGroups(); + if (groups.isEmpty()) { + return; + } + + List fieldNames = schema.getRowType().getFieldNames(); + Set targetNames = new LinkedHashSet<>(); + for (int col : targetColumns) { + if (col >= 0 && col < fieldNames.size()) { + targetNames.add(fieldNames.get(col)); + } + } + + for (Schema.SequenceGroup group : groups) { + Set groupFields = new LinkedHashSet<>(group.getSequenceColumns()); + groupFields.addAll(group.getProtectedColumns()); + + Set missing = new LinkedHashSet<>(); + boolean anyTargeted = false; + for (String field : groupFields) { + if (targetNames.contains(field)) { + anyTargeted = true; + } else { + missing.add(field); + } + } + if (anyTargeted && !missing.isEmpty()) { + throw new InvalidTargetColumnException( + String.format( + "The target write columns must cover the sequence group ordered by %s " + + "entirely or not at all, but %s %s missing.", + group.getSequenceColumns(), + missing, + missing.size() == 1 ? "is" : "are")); + } + } + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java index a7ce4bac9c5..78e025ebdbd 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java @@ -23,6 +23,7 @@ import org.apache.fluss.record.BinaryValue; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.server.kv.rowmerger.SequenceGroups; import org.apache.fluss.types.DataType; import javax.annotation.Nullable; @@ -43,12 +44,30 @@ public class PartialUpdater { private final BitSet primaryKeyCols = new BitSet(); private final boolean updatePrimaryKeyOnly; private final DataType[] fieldDataTypes; + private final @Nullable SequenceGroups sequenceGroups; public PartialUpdater(KvFormat kvFormat, short schemaId, Schema schema, int[] targetColumns) { + this(kvFormat, schemaId, schema, targetColumns, SequenceGroups.create(schema)); + } + + /** + * @param sequenceGroups the sequence groups arbitrating the update, or null to replace the + * target columns blindly as required when recovering by overwriting an already decided + * value + */ + public PartialUpdater( + KvFormat kvFormat, + short schemaId, + Schema schema, + int[] targetColumns, + @Nullable SequenceGroups sequenceGroups) { this.targetSchemaId = schemaId; for (int targetColumn : targetColumns) { partialUpdateCols.set(targetColumn); } + // a group the write doesn't cover must not arbitrate, since its sequence is never stored + this.sequenceGroups = + sequenceGroups == null ? null : sequenceGroups.restrictTo(partialUpdateCols); for (int pkIndex : schema.getPrimaryKeyIndexes()) { primaryKeyCols.set(pkIndex); } @@ -97,6 +116,9 @@ private void sanityCheck(Schema schema, int[] targetColumns) { * oldValue} may be null, in this case, the field don't exist in the {@code partialRow} will be * set to null. * + *

When the schema declares sequence groups, a target column is only taken from {@code + * partialValue} if the group arbitrating it advances, otherwise the stored value is kept. + * * @param oldValue the old value to be updated * @param partialValue the new value to be updated. * @return the updated value (schema id + row bytes) @@ -107,11 +129,23 @@ public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partial return oldValue; } + if (sequenceGroups != null) { + sequenceGroups.arbitrate(oldValue == null ? null : oldValue.row, partialValue.row); + // a fully rejected write is a no-op: return the stored row itself, so the processor + // sees no change. Only under the target schema, since returning it keeps that schema. + if (oldValue != null + && oldValue.schemaId == targetSchemaId + && sequenceGroups.rejectsEveryTargetField(partialUpdateCols, false)) { + return oldValue; + } + } + rowEncoder.startNewRow(); // write each field for (int i = 0; i < fieldDataTypes.length; i++) { - // use the partial row value - if (partialUpdateCols.get(i)) { + // use the partial row value, unless the sequence group arbitrating the field holds it + // back because the incoming row is not newer + if (partialUpdateCols.get(i) && (sequenceGroups == null || sequenceGroups.accepts(i))) { rowEncoder.encodeField(i, flussFieldGetters[i].getFieldOrNull(partialValue.row)); } else { // use the old row value, the old row may be old schema with fewer fields, @@ -136,6 +170,8 @@ public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partial * @return the value after partial deleted */ public @Nullable BinaryValue deleteRow(BinaryValue value) { + // TODO: arbitrate the delete with the sequence groups when a delete record carries the + // sequence columns, so that a stale delete no longer nulls out newer columns if (isFieldsNull(value.row, partialUpdateCols)) { return null; } else { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java index f84769eb2e1..af10bfb9030 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMerger.java @@ -25,6 +25,7 @@ import org.apache.fluss.record.BinaryValue; import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.server.kv.TargetColumns; import org.apache.fluss.server.kv.rowmerger.aggregate.AggregateFieldsProcessor; import org.apache.fluss.server.kv.rowmerger.aggregate.AggregationContext; import org.apache.fluss.server.kv.rowmerger.aggregate.AggregationContextCache; @@ -70,6 +71,10 @@ public class AggregateRowMerger implements RowMerger { // the current target schema id which is updated before merge() operation private short targetSchemaId = -1; + // the all-fields bit set for the current target schema, computed on demand and reused + private @Nullable BitSet allTargetFields; + private int allTargetFieldsCount = -1; + public AggregateRowMerger( TableConfig tableConfig, KvFormat kvFormat, SchemaGetter schemaGetter) { this.schemaGetter = schemaGetter; @@ -86,15 +91,27 @@ public AggregateRowMerger( @Override public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { - // First write: no existing row - if (oldValue == null || oldValue.row == null) { + boolean firstWrite = oldValue == null || oldValue.row == null; + AggregationContext newContext = contextCache.getContext(newValue.schemaId); + AggregationContext targetContext = contextCache.getContext(targetSchemaId); + SequenceGroups sequenceGroups = targetContext.getSequenceGroups(); + if (firstWrite && acceptsEveryField(sequenceGroups, null, newValue.row)) { return newValue; } - // Get contexts for schema evolution support - AggregationContext oldContext = contextCache.getContext(oldValue.schemaId); - AggregationContext newContext = contextCache.getContext(newValue.schemaId); - AggregationContext targetContext = contextCache.getContext(targetSchemaId); + // a fully rejected write is a no-op: return the stored row itself, so the processor sees + // no change. Only under the target schema, since returning it keeps that schema. + if (sequenceGroups != null + && !firstWrite + && oldValue.schemaId == targetSchemaId + && rejectsEveryTargetField( + sequenceGroups, allFields(targetContext), oldValue.row, newValue.row)) { + return oldValue; + } + + // Get the old context only when a stored row exists + AggregationContext oldContext = + firstWrite ? null : contextCache.getContext(oldValue.schemaId); // Use target schema encoder to ensure merged row uses latest schema RowEncoder encoder = targetContext.getRowEncoder(); @@ -102,12 +119,54 @@ public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { // Aggregate using target schema context to ensure output uses server's latest schema AggregateFieldsProcessor.aggregateAllFieldsWithTargetSchema( - oldValue.row, newValue.row, oldContext, newContext, targetContext, encoder); + firstWrite ? null : oldValue.row, + newValue.row, + oldContext, + newContext, + targetContext, + sequenceGroups, + encoder); BinaryRow mergedRow = encoder.finishRow(); return new BinaryValue(targetSchemaId, mergedRow); } + /** + * Returns whether the first row is accepted as it is, i.e. there is no group to arbitrate or + * every group advances. The arbitration result is left in the reused buffer, ready for the + * aggregation path when the row is not accepted. + */ + private static boolean acceptsEveryField( + @Nullable SequenceGroups sequenceGroups, @Nullable BinaryRow oldRow, BinaryRow newRow) { + if (sequenceGroups == null) { + return true; + } + sequenceGroups.arbitrate(oldRow, newRow); + return sequenceGroups.acceptsEveryArbitratedGroup(); + } + + /** Arbitrates and returns whether the write contributes nothing to any target field. */ + private static boolean rejectsEveryTargetField( + SequenceGroups sequenceGroups, + BitSet targetFields, + BinaryRow oldRow, + BinaryRow newRow) { + sequenceGroups.arbitrate(oldRow, newRow); + return sequenceGroups.rejectsEveryTargetField(targetFields, true); + } + + /** Returns a cached bit set covering every field of the given context's schema. */ + private BitSet allFields(AggregationContext context) { + int fieldCount = context.getFieldCount(); + if (allTargetFields == null || allTargetFieldsCount != fieldCount) { + BitSet all = new BitSet(); + all.set(0, fieldCount); + this.allTargetFields = all; + this.allTargetFieldsCount = fieldCount; + } + return allTargetFields; + } + @Override public BinaryValue delete(BinaryValue oldValue) { // Remove the entire row (returns null to indicate deletion) @@ -127,6 +186,8 @@ public RowMerger configureTargetColumns( return this; } + TargetColumns.checkSequenceGroupsAreFullyTargeted(latestSchema, targetColumns); + // Use cache to get or create PartialAggregateRowMerger // This avoids repeated object creation and BitSet construction CacheKey cacheKey = new CacheKey(latestSchemaId, targetColumns); @@ -242,6 +303,12 @@ private static class PartialAggregateRowMerger implements RowMerger { // operations private final Cache targetPosBitSetCache; + // The groups restricted to the target fields, null when the schema declares none + private final @Nullable SequenceGroups sequenceGroups; + + // the target fields as row field indexes, matching the schema this merger was built for + private final BitSet targetFieldPositions; + PartialAggregateRowMerger( BitSet targetColumnIdBitSet, DeleteBehavior deleteBehavior, @@ -259,6 +326,13 @@ private static class PartialAggregateRowMerger implements RowMerger { AggregationContext context = contextCache.getOrCreateContext(schemaId, schema); context.sanityCheckTargetColumns(targetColumnIdBitSet); + // a group the write doesn't cover must not arbitrate, since its sequence is never + // stored + this.targetFieldPositions = targetPositions(schema, targetColumnIdBitSet); + SequenceGroups declared = context.getSequenceGroups(); + this.sequenceGroups = + declared == null ? null : declared.restrictTo(targetFieldPositions); + // Initialize cache for target position BitSets this.targetPosBitSetCache = Caffeine.newBuilder() @@ -267,15 +341,38 @@ private static class PartialAggregateRowMerger implements RowMerger { .build(); } + /** Maps the target column ids onto the row field indexes of the given schema. */ + private static BitSet targetPositions(Schema schema, BitSet targetColumnIdBitSet) { + BitSet positions = new BitSet(); + List columns = schema.getColumns(); + for (int pos = 0; pos < columns.size(); pos++) { + if (targetColumnIdBitSet.get(columns.get(pos).getColumnId())) { + positions.set(pos); + } + } + return positions; + } + @Override public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { - // First write: no existing row - if (oldValue == null || oldValue.row == null) { + boolean firstWrite = oldValue == null || oldValue.row == null; + if (firstWrite && acceptsEveryField(sequenceGroups, null, newValue.row)) { return newValue; } + // the same no-op shortcut as the full aggregation path, over this write's target + // fields; the groups here are already restricted to those fields + if (sequenceGroups != null + && !firstWrite + && oldValue.schemaId == targetSchemaId + && rejectsEveryTargetField( + sequenceGroups, targetFieldPositions, oldValue.row, newValue.row)) { + return oldValue; + } + // Get contexts for schema evolution support - AggregationContext oldContext = contextCache.getContext(oldValue.schemaId); + AggregationContext oldContext = + firstWrite ? null : contextCache.getContext(oldValue.schemaId); AggregationContext newContext = contextCache.getContext(newValue.schemaId); AggregationContext targetContext = contextCache.getContext(targetSchemaId); @@ -285,12 +382,13 @@ public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { // Aggregate using target schema to ensure output uses server's latest schema AggregateFieldsProcessor.aggregateTargetFieldsWithTargetSchema( - oldValue.row, + firstWrite ? null : oldValue.row, newValue.row, oldContext, newContext, targetContext, targetColumnIdBitSet, + sequenceGroups, encoder); BinaryRow mergedRow = encoder.finishRow(); diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMerger.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMerger.java index d7f7eacfdd9..a9605e362eb 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMerger.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMerger.java @@ -21,9 +21,12 @@ import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.Schema; import org.apache.fluss.record.BinaryValue; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.row.encode.RowEncoder; import org.apache.fluss.server.kv.TargetColumns; import org.apache.fluss.server.kv.partialupdate.PartialUpdater; import org.apache.fluss.server.kv.partialupdate.PartialUpdaterCache; +import org.apache.fluss.types.DataType; import javax.annotation.Nullable; @@ -40,15 +43,38 @@ public class DefaultRowMerger implements RowMerger { private final PartialUpdaterCache partialUpdaterCache; private final KvFormat kvFormat; private final DeleteBehavior deleteBehavior; + private final boolean arbitrateSequenceGroups; + + // the full-row merger of the schema resolved last, kept so that a sequence group table doesn't + // rebuild its encoder on every batch. sequence groups only change along with the schema. + private short resolvedSchemaId = -1; + private @Nullable RowMerger sequenceGroupRowMerger; public DefaultRowMerger(KvFormat kvFormat, @Nullable DeleteBehavior deleteBehavior) { + this(kvFormat, deleteBehavior, true); + } + + private DefaultRowMerger( + KvFormat kvFormat, + @Nullable DeleteBehavior deleteBehavior, + boolean arbitrateSequenceGroups) { this.kvFormat = kvFormat; + this.arbitrateSequenceGroups = arbitrateSequenceGroups; // for compatibility, default to ALLOW if not specified this.deleteBehavior = deleteBehavior != null ? deleteBehavior : DeleteBehavior.ALLOW; // TODO: share cache in server level when PartialUpdater is thread-safe this.partialUpdaterCache = new PartialUpdaterCache(); } + /** + * Creates a merger that replaces values blindly, bypassing the sequence groups declared on the + * schema. Used to recover by overwriting an already decided value: such a write restores an + * earlier state, so arbitrating it would reject it as stale and leave the row inconsistent. + */ + public static DefaultRowMerger forBlindOverwrite(KvFormat kvFormat) { + return new DefaultRowMerger(kvFormat, DeleteBehavior.ALLOW, false); + } + @Nullable @Override public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { @@ -73,16 +99,45 @@ public RowMerger configureTargetColumns( @Nullable int[] targetColumns, short latestShemaId, Schema latestSchema) { if (targetColumns == null || TargetColumns.specifiesAllSchemaFieldIndexes(latestSchema, targetColumns)) { - return this; + return fullRowMerger(latestShemaId, latestSchema); } else { + TargetColumns.checkSequenceGroupsAreFullyTargeted(latestSchema, targetColumns); // this also sanity checks the validity of the partial update PartialUpdater partialUpdater = - partialUpdaterCache.getOrCreatePartialUpdater( - kvFormat, latestShemaId, latestSchema, targetColumns); + arbitrateSequenceGroups + ? partialUpdaterCache.getOrCreatePartialUpdater( + kvFormat, latestShemaId, latestSchema, targetColumns) + : new PartialUpdater( + kvFormat, latestShemaId, latestSchema, targetColumns, null); return new PartialUpdateRowMerger(partialUpdater, deleteBehavior); } } + /** + * Returns the merger handling a full-row write. Without sequence groups the new row always + * wins, so this merger is used as it is; with sequence groups the stored row has to be + * consulted to arbitrate every group, which needs a merger of its own. + */ + private RowMerger fullRowMerger(short latestSchemaId, Schema latestSchema) { + if (!arbitrateSequenceGroups) { + return this; + } + if (latestSchemaId != resolvedSchemaId) { + SequenceGroups sequenceGroups = SequenceGroups.create(latestSchema); + sequenceGroupRowMerger = + sequenceGroups == null + ? null + : new SequenceGroupRowMerger( + kvFormat, + latestSchemaId, + latestSchema, + sequenceGroups, + deleteBehavior); + resolvedSchemaId = latestSchemaId; + } + return sequenceGroupRowMerger == null ? this : sequenceGroupRowMerger; + } + /** A merger that partially updates specified columns with the new row. */ private static class PartialUpdateRowMerger implements RowMerger { @@ -119,4 +174,87 @@ public DeleteBehavior deleteBehavior() { return deleteBehavior; } } + + /** + * A merger that arbitrates a full-row write with sequence groups: a column only takes the + * incoming value if the group protecting it advances, otherwise the stored value survives. + * Since this engine has no aggregate functions, a group that doesn't advance simply drops the + * incoming values, so the outcome depends only on the largest sequence seen per group rather + * than on the order the records arrive in. + * + *

Sequence groups arbitrate writes only: a delete carries no sequence values to compare + * against the stored row, so it keeps removing the whole row as it did before sequence groups + * existed. + */ + private static class SequenceGroupRowMerger implements RowMerger { + + private final SequenceGroups sequenceGroups; + private final InternalRow.FieldGetter[] fieldGetters; + private final RowEncoder rowEncoder; + private final short targetSchemaId; + private final DeleteBehavior deleteBehavior; + + SequenceGroupRowMerger( + KvFormat kvFormat, + short targetSchemaId, + Schema schema, + SequenceGroups sequenceGroups, + DeleteBehavior deleteBehavior) { + this.sequenceGroups = sequenceGroups; + this.targetSchemaId = targetSchemaId; + this.deleteBehavior = deleteBehavior; + DataType[] fieldDataTypes = schema.getRowType().getChildren().toArray(new DataType[0]); + this.fieldGetters = new InternalRow.FieldGetter[fieldDataTypes.length]; + for (int i = 0; i < fieldDataTypes.length; i++) { + fieldGetters[i] = InternalRow.createFieldGetter(fieldDataTypes[i], i); + } + this.rowEncoder = RowEncoder.create(kvFormat, fieldDataTypes); + } + + @Nullable + @Override + public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { + sequenceGroups.arbitrate(oldValue == null ? null : oldValue.row, newValue.row); + if (sequenceGroups.acceptsEveryArbitratedGroup()) { + // Every group advances, so the whole incoming row wins + return newValue; + } + + rowEncoder.startNewRow(); + for (int i = 0; i < fieldGetters.length; i++) { + InternalRow source = + sequenceGroups.accepts(i) + ? newValue.row + : oldValue == null ? null : oldValue.row; + // the stored row may be absent or follow an older schema with fewer fields, in + // which case the missing fields are null + if (source == null || source.getFieldCount() < i + 1) { + rowEncoder.encodeField(i, null); + } else { + rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(source)); + } + } + return new BinaryValue(targetSchemaId, rowEncoder.finishRow()); + } + + @Nullable + @Override + public BinaryValue delete(BinaryValue oldRow) { + // TODO: arbitrate the delete with the sequence groups when a delete record carries the + // sequence columns, so that a stale delete no longer drops a newer row + return null; + } + + @Override + public DeleteBehavior deleteBehavior() { + return deleteBehavior; + } + + @Override + public RowMerger configureTargetColumns( + @Nullable int[] targetColumns, short schemaId, Schema schema) { + throw new IllegalStateException( + "SequenceGroupRowMerger does not support reconfigure target merge columns."); + } + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/SequenceGroups.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/SequenceGroups.java new file mode 100644 index 00000000000..1f8871c0455 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/SequenceGroups.java @@ -0,0 +1,404 @@ +/* + * 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.fluss.server.kv.rowmerger; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.LocalZonedTimestampType; +import org.apache.fluss.types.RowType; +import org.apache.fluss.types.TimestampType; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.util.Arrays; +import java.util.BitSet; +import java.util.List; + +import static org.apache.fluss.utils.Preconditions.checkArgument; + +/** + * The sequence groups declared on a schema, resolved into field positions so that a merger can + * arbitrate each group on its own. + * + *

A column is put under the order of one or more sequence columns (see {@link + * Schema.SequenceGroup#getSequenceColumns()}) and then only takes an incoming value when those + * sequence columns are not older than the stored ones. Columns ordered by the very same sequence + * columns form one group advancing together, while different groups advance independently: within a + * single write one group may advance and another may not. That is what distinguishes sequence + * groups from the versioned merge engine, which arbitrates the whole row with a single version. + * + *

One instance serves all keys of a table. The group decisions are reused per record, safe + * because the write path is single threaded under KvTablet's write lock. + */ +@Internal +public class SequenceGroups implements Serializable { + + private static final long serialVersionUID = 1L; + + /** + * What a group makes of an incoming row, once its sequence columns have been compared with the + * stored ones. + * + *

A merger without aggregate functions treats {@link #SKIP} and {@link #STALE} alike, since + * both keep the stored values. One with aggregate functions has to tell them apart: a skipped + * group contributes nothing at all, while a stale one still aggregates, only as a record that + * happened earlier. + */ + public enum Decision { + /** + * The incoming row carries no sequence value for the group at all, so the group has no way + * to order it and leaves its fields untouched. + */ + SKIP, + + /** + * The incoming sequence is not older than the stored one, so the group moves forward: its + * fields take the incoming values and its sequence columns advance along with them. + */ + FORWARD, + + /** + * The incoming sequence is older than the stored one. The group doesn't move forward, so + * its sequence columns keep the stored values, and an aggregate function sees the incoming + * row as one that happened earlier. + */ + STALE + } + + /** A field taking part in no group, so it is never held back. */ + private static final int NO_GROUP = -1; + + /** + * For each field, the group arbitrating it, or {@link #NO_GROUP} if the field takes part in no + * group. A sequence column is arbitrated by the very group it orders, so that the whole group + * advances at once. + */ + private final int[] groupOfField; + + /** + * For each group, the field indexes of its sequence columns, in the declared comparison order. + */ + private final int[][] sequenceFieldsOfGroup; + + /** + * For each group, the comparators of its sequence columns, matching {@link + * #sequenceFieldsOfGroup}. They compare non-null values only, so that null ordering is decided + * once in {@link #decide} instead of per column type. + */ + private final SequenceComparator[][] comparatorsOfGroup; + + /** Stands for a group left out of the arbitration by {@link #restrictTo}. */ + private static final int[] NO_FIELDS = {}; + + /** The decision of every group for the record under arbitration, indexed by group id. */ + private final Decision[] groupDecisions; + + /** The primary key field indexes, which hold the same value in both rows being merged. */ + private final BitSet primaryKeyFields; + + private SequenceGroups( + int[] groupOfField, + int[][] sequenceFieldsOfGroup, + SequenceComparator[][] comparatorsOfGroup, + BitSet primaryKeyFields) { + this.groupOfField = groupOfField; + this.sequenceFieldsOfGroup = sequenceFieldsOfGroup; + this.comparatorsOfGroup = comparatorsOfGroup; + this.groupDecisions = new Decision[comparatorsOfGroup.length]; + Arrays.fill(groupDecisions, Decision.FORWARD); + this.primaryKeyFields = primaryKeyFields; + } + + /** + * Resolves the sequence groups of the given schema, or returns null if the schema declares + * none. Returning null lets callers keep their original merge path untouched. + */ + @Nullable + public static SequenceGroups create(Schema schema) { + List declared = schema.getSequenceGroups(); + if (declared.isEmpty()) { + return null; + } + + RowType rowType = schema.getRowType(); + int fieldCount = rowType.getFieldCount(); + + int[] groupOfField = new int[fieldCount]; + Arrays.fill(groupOfField, NO_GROUP); + + int[][] sequenceFieldsOfGroup = new int[declared.size()][]; + SequenceComparator[][] comparatorsOfGroup = new SequenceComparator[declared.size()][]; + for (int groupId = 0; groupId < declared.size(); groupId++) { + Schema.SequenceGroup group = declared.get(groupId); + List sequenceColumns = group.getSequenceColumns(); + int[] sequenceFields = new int[sequenceColumns.size()]; + SequenceComparator[] comparators = new SequenceComparator[sequenceColumns.size()]; + for (int i = 0; i < sequenceColumns.size(); i++) { + String sequenceColumn = sequenceColumns.get(i); + int sequenceField = rowType.getFieldIndex(sequenceColumn); + checkArgument( + sequenceField >= 0, + "The sequence column '%s' doesn't exist in schema.", + sequenceColumn); + sequenceFields[i] = sequenceField; + comparators[i] = + createComparator( + sequenceColumn, rowType.getTypeAt(sequenceField), sequenceField); + // a sequence column takes part in the very group it orders, otherwise it would + // always accept incoming values and report a sequence no longer matching them + groupOfField[sequenceField] = groupId; + } + sequenceFieldsOfGroup[groupId] = sequenceFields; + comparatorsOfGroup[groupId] = comparators; + + for (String protectedColumn : group.getProtectedColumns()) { + int fieldIndex = rowType.getFieldIndex(protectedColumn); + checkArgument( + fieldIndex >= 0, + "The protected column '%s' doesn't exist in schema.", + protectedColumn); + groupOfField[fieldIndex] = groupId; + } + } + + BitSet primaryKeyFields = new BitSet(); + for (int pkIndex : schema.getPrimaryKeyIndexes()) { + primaryKeyFields.set(pkIndex); + } + return new SequenceGroups( + groupOfField, sequenceFieldsOfGroup, comparatorsOfGroup, primaryKeyFields); + } + + /** + * Returns the groups arbitrating only the fields the target set covers, since a group whose + * sequence is never stored must not decide on values the row keeps. + * + * @param targetFields the row field indexes the write targets + */ + public SequenceGroups restrictTo(BitSet targetFields) { + int[] restricted = groupOfField.clone(); + for (int i = 0; i < restricted.length; i++) { + if (!targetFields.get(i)) { + restricted[i] = NO_GROUP; + } + } + + int[][] restrictedFields = sequenceFieldsOfGroup.clone(); + for (int groupId = 0; groupId < restrictedFields.length; groupId++) { + if (!coversGroup(restricted, groupId)) { + restrictedFields[groupId] = NO_FIELDS; + } + } + return new SequenceGroups( + restricted, restrictedFields, comparatorsOfGroup, primaryKeyFields); + } + + /** Returns whether any field still belongs to the given group. */ + private static boolean coversGroup(int[] groupOfField, int groupId) { + for (int owner : groupOfField) { + if (owner == groupId) { + return true; + } + } + return false; + } + + /** + * Decides every covered group for the incoming row, into the reused decision buffer. The + * decisions live until the next arbitration. + * + * @param oldRow the stored row, or null when there is no stored row yet + * @param newRow the incoming row + */ + public void arbitrate(@Nullable InternalRow oldRow, InternalRow newRow) { + for (int groupId = 0; groupId < comparatorsOfGroup.length; groupId++) { + groupDecisions[groupId] = + decide( + sequenceFieldsOfGroup[groupId], + comparatorsOfGroup[groupId], + oldRow, + newRow); + } + } + + /** + * Returns whether the field may take the value carried by the last arbitrated row. A field is + * held back only when the group arbitrating it doesn't advance; without aggregate functions a + * skipped group and a stale one both keep the stored values, so the two need no telling apart + * here. + */ + public boolean accepts(int fieldIndex) { + int groupId = groupOfField[fieldIndex]; + return groupId == NO_GROUP || groupDecisions[groupId] == Decision.FORWARD; + } + + /** Returns the field count of the schema these groups were resolved from. */ + public int fieldCount() { + return groupOfField.length; + } + + /** + * Returns what the group arbitrating the field makes of the last arbitrated row. A field taking + * part in no group always reports {@link Decision#FORWARD}, keeping its original behavior; + * callers that aggregate use this rather than {@link #accepts}, so that they can aggregate a + * stale row in reverse instead of dropping it. + */ + public Decision decisionOf(int fieldIndex) { + int groupId = groupOfField[fieldIndex]; + return groupId == NO_GROUP ? Decision.FORWARD : groupDecisions[groupId]; + } + + /** Returns whether every arbitrated group advances. */ + public boolean acceptsEveryArbitratedGroup() { + for (Decision decision : groupDecisions) { + if (decision != Decision.FORWARD) { + return false; + } + } + return true; + } + + /** + * Returns whether every target field of the write is rejected, so the write changes nothing. A + * target field is either a primary key, holding the same value in both rows, or arbitrated by a + * group that rejects the incoming value; a field outside the groups takes the incoming value + * unconditionally, so it counts as a contribution. + * + *

Without aggregate functions SKIP and STALE alike keep the stored values, while an + * aggregating engine still folds a stale record in through aggReversed, so there only SKIP + * rejects. + * + * @param aggregating whether the merging engine aggregates, i.e. whether a stale record still + * contributes + */ + public boolean rejectsEveryTargetField(BitSet targetFields, boolean aggregating) { + for (int i = targetFields.nextSetBit(0); i >= 0; i = targetFields.nextSetBit(i + 1)) { + if (primaryKeyFields.get(i)) { + continue; + } + if (groupOfField[i] == NO_GROUP) { + return false; + } + Decision decision = groupDecisions[groupOfField[i]]; + if (decision == Decision.FORWARD || (aggregating && decision == Decision.STALE)) { + return false; + } + } + return true; + } + + /** + * Decides one group, by comparing its sequence columns in the declared order until one of them + * differs. The values are compared column by column and never stored, so deciding allocates + * nothing and never boxes a sequence value. + */ + private static Decision decide( + int[] sequenceFields, + SequenceComparator[] comparators, + @Nullable InternalRow oldRow, + InternalRow newRow) { + boolean carriesValue = false; + for (int field : sequenceFields) { + if (!absent(newRow, field)) { + carriesValue = true; + break; + } + } + if (!carriesValue) { + // the group carries no order information at all + return Decision.SKIP; + } + if (oldRow == null) { + return Decision.FORWARD; + } + + for (int i = 0; i < sequenceFields.length; i++) { + int field = sequenceFields[i]; + // SQL NULL orders before every value, and a column absent from an older schema is null + int comparison; + if (absent(newRow, field)) { + comparison = absent(oldRow, field) ? 0 : -1; + } else if (absent(oldRow, field)) { + comparison = 1; + } else { + comparison = comparators[i].compareNonNull(oldRow, newRow); + } + if (comparison != 0) { + return comparison > 0 ? Decision.FORWARD : Decision.STALE; + } + } + // equal sequences advance, so that a replayed record still refreshes the group + return Decision.FORWARD; + } + + /** + * Returns a comparator of the given sequence column, and validates that its type can order a + * group. The accepted types are the same as the version column of the versioned merge engine, + * so that both order arbitration mechanisms stay consistent. + */ + private static SequenceComparator createComparator( + String columnName, DataType dataType, int fieldIndex) { + switch (dataType.getTypeRoot()) { + case INTEGER: + return (oldRow, newRow) -> + Integer.compare(newRow.getInt(fieldIndex), oldRow.getInt(fieldIndex)); + case BIGINT: + return (oldRow, newRow) -> + Long.compare(newRow.getLong(fieldIndex), oldRow.getLong(fieldIndex)); + case TIMESTAMP_WITHOUT_TIME_ZONE: + int ntzPrecision = ((TimestampType) dataType).getPrecision(); + return (oldRow, newRow) -> + newRow.getTimestampNtz(fieldIndex, ntzPrecision) + .compareTo(oldRow.getTimestampNtz(fieldIndex, ntzPrecision)); + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + int ltzPrecision = ((LocalZonedTimestampType) dataType).getPrecision(); + return (oldRow, newRow) -> + newRow.getTimestampLtz(fieldIndex, ltzPrecision) + .compareTo(oldRow.getTimestampLtz(fieldIndex, ltzPrecision)); + default: + throw new IllegalArgumentException( + String.format( + "The sequence column '%s' must be one type of " + + "[INT, BIGINT, TIMESTAMP, TIMESTAMP_LTZ], but is %s.", + columnName, dataType)); + } + } + + /** + * A row written under an older schema may carry fewer fields than the latest schema, in which + * case the sequence column is absent and read as null, i.e. the oldest sequence. + */ + private static boolean absent(InternalRow row, int fieldIndex) { + return row.getFieldCount() < fieldIndex + 1 || row.isNullAt(fieldIndex); + } + + /** + * Compares the sequence value of one column between the stored and the incoming row. Both + * values are known to be non-null; the null ordering lives in {@link #decide} so it is decided + * once instead of per column type. + */ + @FunctionalInterface + private interface SequenceComparator extends Serializable { + + /** Returns a negative number when the incoming value is older than the stored one. */ + int compareNonNull(InternalRow oldRow, InternalRow newRow); + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java index 0429a2015bb..8252f264a35 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregateFieldsProcessor.java @@ -21,8 +21,11 @@ import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.server.kv.rowmerger.SequenceGroups; import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldAggregator; +import javax.annotation.Nullable; + import java.util.BitSet; import java.util.List; @@ -54,28 +57,38 @@ private AggregateFieldsProcessor() {} *

This method handles schema evolution by matching fields using column IDs across three * potentially different schemas: old row schema, new row schema, and target output schema. * - * @param oldRow the old row + * @param oldRow the old row, or null when there is no stored row yet * @param newRow the new row - * @param oldContext context for the old row schema + * @param oldContext context for the old row schema, or null when there is no stored row yet * @param newInputContext context for the new row schema (for reading newRow) * @param targetContext context for the target output schema + * @param sequenceGroups the sequence groups arbitrating the merge, or null when the schema + * declares none * @param encoder the row encoder to encode results (should match targetContext) */ public static void aggregateAllFieldsWithTargetSchema( - BinaryRow oldRow, + @Nullable BinaryRow oldRow, BinaryRow newRow, - AggregationContext oldContext, + @Nullable AggregationContext oldContext, AggregationContext newInputContext, AggregationContext targetContext, + @Nullable SequenceGroups sequenceGroups, RowEncoder encoder) { + // the groups are resolved against the target schema, which is the one being encoded + if (sequenceGroups != null) { + sequenceGroups.arbitrate(oldRow, newRow); + } + // Fast path: all three schemas are the same - if (targetContext == oldContext && targetContext == newInputContext) { - aggregateAllFieldsWithSameSchema(oldRow, newRow, targetContext, encoder); + if (oldRow != null && targetContext == oldContext && targetContext == newInputContext) { + aggregateAllFieldsWithSameSchema( + oldRow, newRow, targetContext, sequenceGroups, encoder); return; } // General path: iterate over target schema columns and aggregate using column ID matching - InternalRow.FieldGetter[] oldFieldGetters = oldContext.getFieldGetters(); + InternalRow.FieldGetter[] oldFieldGetters = + oldContext == null ? null : oldContext.getFieldGetters(); InternalRow.FieldGetter[] newFieldGetters = newInputContext.getFieldGetters(); FieldAggregator[] targetAggregators = targetContext.getAggregators(); List targetColumns = targetContext.getSchema().getColumns(); @@ -85,7 +98,7 @@ public static void aggregateAllFieldsWithTargetSchema( int columnId = targetColumn.getColumnId(); // Find corresponding fields in old and new schemas using column ID - Integer oldIdx = oldContext.getFieldIndex(columnId); + Integer oldIdx = oldContext == null ? null : oldContext.getFieldIndex(columnId); Integer newIdx = newInputContext.getFieldIndex(columnId); // Get field getters (use NULL_FIELD_GETTER if column doesn't exist in that schema) @@ -101,33 +114,58 @@ public static void aggregateAllFieldsWithTargetSchema( oldRow, newRow, targetAggregators[targetIdx], + decisionOf(sequenceGroups, targetIdx), targetIdx, encoder); } } + /** + * Returns what the group arbitrating the given field makes of the incoming row, defaulting to + * {@link SequenceGroups.Decision#FORWARD} when the schema declares no sequence group at all. + */ + private static SequenceGroups.Decision decisionOf( + @Nullable SequenceGroups sequenceGroups, int fieldIndex) { + return sequenceGroups == null + ? SequenceGroups.Decision.FORWARD + : sequenceGroups.decisionOf(fieldIndex); + } + /** * Aggregate and encode a single field. * * @param oldFieldGetter getter for the old field * @param newFieldGetter getter for the new field - * @param oldRow the old row + * @param oldRow the old row, or null when there is no stored row yet * @param newRow the new row * @param aggregator the aggregator for this field + * @param decision what the sequence group arbitrating this field makes of the incoming row, or + * {@link SequenceGroups.Decision#FORWARD} when no group arbitrates it * @param targetIdx the target index to encode * @param encoder the row encoder */ private static void aggregateAndEncode( InternalRow.FieldGetter oldFieldGetter, InternalRow.FieldGetter newFieldGetter, - BinaryRow oldRow, + @Nullable BinaryRow oldRow, BinaryRow newRow, FieldAggregator aggregator, + SequenceGroups.Decision decision, int targetIdx, RowEncoder encoder) { - Object accumulator = oldFieldGetter.getFieldOrNull(oldRow); + Object accumulator = oldRow == null ? null : oldFieldGetter.getFieldOrNull(oldRow); + if (decision == SequenceGroups.Decision.SKIP) { + // the incoming row carries no sequence for the group, so it contributes nothing + encoder.encodeField(targetIdx, accumulator); + return; + } + Object inputField = newFieldGetter.getFieldOrNull(newRow); - Object mergedField = aggregator.agg(accumulator, inputField); + // a stale row still aggregates, only as one that happened before the stored value + Object mergedField = + decision == SequenceGroups.Decision.STALE + ? aggregator.aggReversed(accumulator, inputField) + : aggregator.agg(accumulator, inputField); encoder.encodeField(targetIdx, mergedField); } @@ -159,31 +197,39 @@ private static void copyOldValueAndEncode( * the old value unchanged. For columns that don't exist in old schema, copy from newRow. For * columns that exist only in target schema, set to null. * - * @param oldRow the old row + * @param oldRow the old row, or null when there is no stored row yet * @param newRow the new row - * @param oldContext context for the old row schema + * @param oldContext context for the old row schema, or null when there is no stored row yet * @param newInputContext context for the new row schema (for reading newRow) * @param targetContext context for the target output schema * @param targetColumnIdBitSet BitSet marking target columns by column ID + * @param sequenceGroups the sequence groups arbitrating the merge, or null when the schema + * declares none * @param encoder the row encoder to encode results (should match targetContext) */ public static void aggregateTargetFieldsWithTargetSchema( - BinaryRow oldRow, + @Nullable BinaryRow oldRow, BinaryRow newRow, - AggregationContext oldContext, + @Nullable AggregationContext oldContext, AggregationContext newInputContext, AggregationContext targetContext, BitSet targetColumnIdBitSet, + @Nullable SequenceGroups sequenceGroups, RowEncoder encoder) { + if (sequenceGroups != null) { + sequenceGroups.arbitrate(oldRow, newRow); + } + // Fast path: all three schemas are the same - if (targetContext == oldContext && targetContext == newInputContext) { + if (oldRow != null && targetContext == oldContext && targetContext == newInputContext) { aggregateTargetFieldsWithSameSchema( - oldRow, newRow, targetContext, targetColumnIdBitSet, encoder); + oldRow, newRow, targetContext, targetColumnIdBitSet, sequenceGroups, encoder); return; } // General path: iterate over target schema columns - InternalRow.FieldGetter[] oldFieldGetters = oldContext.getFieldGetters(); + InternalRow.FieldGetter[] oldFieldGetters = + oldContext == null ? null : oldContext.getFieldGetters(); InternalRow.FieldGetter[] newFieldGetters = newInputContext.getFieldGetters(); FieldAggregator[] targetAggregators = targetContext.getAggregators(); List targetColumns = targetContext.getSchema().getColumns(); @@ -193,7 +239,7 @@ public static void aggregateTargetFieldsWithTargetSchema( int columnId = targetColumn.getColumnId(); // Find corresponding fields in old and new schemas using column ID - Integer oldIdx = oldContext.getFieldIndex(columnId); + Integer oldIdx = oldContext == null ? null : oldContext.getFieldIndex(columnId); Integer newIdx = newInputContext.getFieldIndex(columnId); if (targetColumnIdBitSet.get(columnId)) { @@ -208,6 +254,7 @@ public static void aggregateTargetFieldsWithTargetSchema( oldRow, newRow, targetAggregators[targetIdx], + decisionOf(sequenceGroups, targetIdx), targetIdx, encoder); } else if (oldIdx != null) { @@ -229,7 +276,11 @@ public static void aggregateTargetFieldsWithTargetSchema( *

Fast path: field positions match directly, no column ID lookup needed. */ private static void aggregateAllFieldsWithSameSchema( - BinaryRow oldRow, BinaryRow newRow, AggregationContext context, RowEncoder encoder) { + BinaryRow oldRow, + BinaryRow newRow, + AggregationContext context, + @Nullable SequenceGroups sequenceGroups, + RowEncoder encoder) { InternalRow.FieldGetter[] fieldGetters = context.getFieldGetters(); FieldAggregator[] aggregators = context.getAggregators(); int fieldCount = context.getFieldCount(); @@ -241,6 +292,7 @@ private static void aggregateAllFieldsWithSameSchema( oldRow, newRow, aggregators[idx], + decisionOf(sequenceGroups, idx), idx, encoder); } @@ -256,6 +308,7 @@ private static void aggregateTargetFieldsWithSameSchema( BinaryRow newRow, AggregationContext context, BitSet targetColumnIdBitSet, + @Nullable SequenceGroups sequenceGroups, RowEncoder encoder) { InternalRow.FieldGetter[] fieldGetters = context.getFieldGetters(); FieldAggregator[] aggregators = context.getAggregators(); @@ -273,6 +326,7 @@ private static void aggregateTargetFieldsWithSameSchema( oldRow, newRow, aggregators[idx], + decisionOf(sequenceGroups, idx), idx, encoder); } else { diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java index 7cec63a4e8d..6ae09df8b31 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/AggregationContext.java @@ -25,15 +25,20 @@ import org.apache.fluss.metadata.Schema; import org.apache.fluss.row.InternalRow; import org.apache.fluss.row.encode.RowEncoder; +import org.apache.fluss.server.kv.rowmerger.SequenceGroups; import org.apache.fluss.server.kv.rowmerger.aggregate.factory.FieldAggregatorFactory; import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldAggregator; import org.apache.fluss.types.DataType; import org.apache.fluss.types.RowType; +import javax.annotation.Nullable; + import java.util.BitSet; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; /** * Context for aggregation operations, containing field getters, aggregators, and encoder for a @@ -53,6 +58,12 @@ public class AggregationContext { final RowEncoder rowEncoder; final int fieldCount; + /** + * The sequence groups declared on this schema, or null when it declares none. Resolved once per + * schema, since the groups only change along with the schema itself. + */ + private final @Nullable SequenceGroups sequenceGroups; + /** * Mapping from column ID to field index in this schema. This is used for schema evolution to * correctly match fields between old and new schemas. @@ -70,12 +81,14 @@ private AggregationContext( RowType rowType, InternalRow.FieldGetter[] fieldGetters, FieldAggregator[] aggregators, - RowEncoder rowEncoder) { + RowEncoder rowEncoder, + @Nullable SequenceGroups sequenceGroups) { this.schema = schema; this.rowType = rowType; this.fieldGetters = fieldGetters; this.aggregators = aggregators; this.rowEncoder = rowEncoder; + this.sequenceGroups = sequenceGroups; this.fieldCount = rowType.getFieldCount(); // Build columnId to index mapping for schema evolution support @@ -108,6 +121,14 @@ public FieldAggregator[] getAggregators() { return aggregators; } + /** + * Gets the sequence groups declared on this schema, or null when it declares none. A null + * result lets a caller keep aggregating every field unconditionally. + */ + public @Nullable SequenceGroups getSequenceGroups() { + return sequenceGroups; + } + public int getFieldCount() { return fieldCount; } @@ -224,7 +245,13 @@ public static AggregationContext create(Schema schema, KvFormat kvFormat) { // Create row encoder RowEncoder rowEncoder = RowEncoder.create(kvFormat, rowType); - return new AggregationContext(schema, rowType, fieldGetters, aggregators, rowEncoder); + return new AggregationContext( + schema, + rowType, + fieldGetters, + aggregators, + rowEncoder, + SequenceGroups.create(schema)); } /** @@ -239,6 +266,7 @@ public static AggregationContext create(Schema schema, KvFormat kvFormat) { private static FieldAggregator[] createAggregators(Schema schema) { RowType rowType = schema.getRowType(); List primaryKeys = schema.getPrimaryKeyColumnNames(); + Set sequenceColumns = sequenceColumnNames(schema); List fieldNames = rowType.getFieldNames(); int fieldCount = rowType.getFieldCount(); @@ -249,7 +277,7 @@ private static FieldAggregator[] createAggregators(Schema schema) { DataType fieldType = rowType.getTypeAt(i); // Get the aggregate function for this field - AggFunction aggFunc = getAggFunction(fieldName, primaryKeys, schema); + AggFunction aggFunc = getAggFunction(fieldName, primaryKeys, sequenceColumns, schema); // Get the factory for this aggregation function type and create the aggregator AggFunctionType type = aggFunc.getType(); @@ -273,24 +301,44 @@ private static FieldAggregator[] createAggregators(Schema schema) { * *

    *
  1. Primary key fields use "last_value" (no aggregation) + *
  2. A sequence column uses "last_value" as well, since the group it orders decides when it + * advances and aggregating it would let a stale row move the sequence backwards *
  3. Schema.getAggFunction() - aggregation function defined in Schema (from Column) *
  4. Final fallback: "last_value_ignore_nulls" *
* * @param fieldName the field name * @param primaryKeys the list of primary key field names + * @param sequenceColumns the names of the columns that order a sequence group * @param schema the Schema object * @return the aggregate function to use */ private static AggFunction getAggFunction( - String fieldName, List primaryKeys, Schema schema) { + String fieldName, + List primaryKeys, + Set sequenceColumns, + Schema schema) { // 1. Primary key fields don't aggregate if (primaryKeys.contains(fieldName)) { return AggFunctions.of(AggFunctionType.LAST_VALUE); } - // 2. Check Schema for aggregation function, or use default fallback + // 2. A sequence column is driven by its own group rather than by an aggregate function + if (sequenceColumns.contains(fieldName)) { + return AggFunctions.of(AggFunctionType.LAST_VALUE); + } + + // 3. Check Schema for aggregation function, or use default fallback return schema.getAggFunction(fieldName).orElseGet(AggFunctions::LAST_VALUE_IGNORE_NULLS); } + + /** Collects every column that orders a sequence group of the schema. */ + private static Set sequenceColumnNames(Schema schema) { + Set names = new HashSet<>(); + for (Schema.SequenceGroup group : schema.getSequenceGroups()) { + names.addAll(group.getSequenceColumns()); + } + return names; + } } diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstNonNullValueAgg.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstNonNullValueAgg.java index 5cafd87e8ee..7d6d2ce1681 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstNonNullValueAgg.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstNonNullValueAgg.java @@ -24,7 +24,14 @@ import org.apache.fluss.types.DataType; -/** First non-null value aggregator - keeps the first seen non-null value. */ +/** + * First non-null value aggregator - keeps the first seen non-null value. + * + *

Under a sequence group this aggregator follows the arrival order once a second stale record + * lands: each stale record replaces the previous stale result, so the winner is the stale record + * that arrived last rather than the one with the smallest sequence. With a single stale record the + * result is correct. + */ public class FieldFirstNonNullValueAgg extends FieldAggregator { private static final long serialVersionUID = 1L; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstValueAgg.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstValueAgg.java index 7f281f81766..d97ab8862c1 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstValueAgg.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldFirstValueAgg.java @@ -24,7 +24,14 @@ import org.apache.fluss.types.DataType; -/** First value aggregator - keeps the first seen value. */ +/** + * First value aggregator - keeps the first seen value. + * + *

Under a sequence group this aggregator follows the arrival order once a second stale record + * lands: each stale record replaces the previous stale result, so the winner is the stale record + * that arrived last rather than the one with the smallest sequence. With a single stale record the + * result is correct. + */ public class FieldFirstValueAgg extends FieldAggregator { private static final long serialVersionUID = 1L; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldListaggAgg.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldListaggAgg.java index 138f284de8a..139be4705b6 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldListaggAgg.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/aggregate/functions/FieldListaggAgg.java @@ -26,7 +26,13 @@ import org.apache.fluss.types.StringType; import org.apache.fluss.utils.BinaryStringUtils; -/** List aggregation aggregator - concatenates string values with a delimiter. */ +/** + * List aggregation aggregator - concatenates string values with a delimiter. + * + *

Under a sequence group this aggregator follows the arrival order once a second stale record + * lands: every stale record is prepended, so the stale values end up ordered by arrival rather than + * by sequence. With a single stale record the result is correct. + */ public class FieldListaggAgg extends FieldAggregator { private static final long serialVersionUID = 1L; diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java index 3255fd87cd6..2b51a90ff37 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/TableDescriptorValidation.java @@ -402,6 +402,7 @@ private static void checkMergeEngine( if (mergeEngine != MergeEngineType.AGGREGATION) { validateNoAggregationFunctions(schema); } + validateSequenceGroups(mergeEngine, hasPrimaryKey, schema); if (mergeEngine != null) { if (!hasPrimaryKey) { throw new InvalidConfigException( @@ -456,6 +457,33 @@ private static void checkMergeEngine( } } + /** + * Validates the sequence groups declared on the schema. + * + *

A sequence group puts one or more columns under the order of a sequence column, so that + * each group decides on its own whether an incoming write is newer than the stored row. This + * lets several writers update disjoint column groups of the same row without overwriting each + * other with stale values. + */ + private static void validateSequenceGroups( + @Nullable MergeEngineType mergeEngine, boolean hasPrimaryKey, Schema schema) { + if (schema.getSequenceGroups().isEmpty()) { + return; + } + + // only a primary key table without merge engine, or with the aggregation one, consults + // the sequence groups when merging; schema-level invariants already ran at build time + if (!hasPrimaryKey) { + throw new InvalidConfigException( + "Sequence group is only supported in primary key table."); + } + if (mergeEngine != null && mergeEngine != MergeEngineType.AGGREGATION) { + throw new InvalidConfigException( + String.format( + "Sequence group is not supported for '%s' merge engine.", mergeEngine)); + } + } + /** Validates that the schema doesn't contain any aggregation functions. */ private static void validateNoAggregationFunctions(Schema schema) { for (Schema.Column column : schema.getColumns()) { diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/TargetColumnsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/TargetColumnsTest.java index 0d630b9f9cf..c96304c1099 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/TargetColumnsTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/TargetColumnsTest.java @@ -17,12 +17,16 @@ package org.apache.fluss.server.kv; +import org.apache.fluss.exception.InvalidTargetColumnException; import org.apache.fluss.metadata.Schema; import org.apache.fluss.types.DataTypes; import org.junit.jupiter.api.Test; +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link TargetColumns}. */ @@ -72,4 +76,67 @@ void testTargetColumns() { .isInstanceOf(NullPointerException.class) .hasMessageContaining("schema"); } + + /** Fields are {@code k=0, status=1, amount=2, ts=3, note=4}; {@code ts} orders the group. */ + private static final Schema SEQUENCE_GROUP_SCHEMA = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("status", DataTypes.STRING()) + .column("amount", DataTypes.INT()) + .column("ts", DataTypes.INT()) + .column("note", DataTypes.STRING()) + .sequenceGroup(singletonList("ts"), asList("status", "amount")) + .primaryKey("k") + .build(); + + private static void check(int... targetColumns) { + TargetColumns.checkSequenceGroupsAreFullyTargeted(SEQUENCE_GROUP_SCHEMA, targetColumns); + } + + @Test + void testGroupFullyTargetedOrFullyLeftOutIsAccepted() { + // k, note: the group is left out entirely + assertThatCode(() -> check(0, 4)).doesNotThrowAnyException(); + // k, status, amount, ts: the group is covered entirely + assertThatCode(() -> check(0, 1, 2, 3)).doesNotThrowAnyException(); + assertThatCode(() -> check(0, 1, 2, 3, 4)).doesNotThrowAnyException(); + } + + @Test + void testProtectedColumnLeftOutIsRejected() { + // status and ts move forward while amount would keep a value from the stored sequence + assertThatThrownBy(() -> check(0, 1, 3)) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "The target write columns must cover the sequence group ordered by [ts] " + + "entirely or not at all, but [amount] is missing."); + } + + @Test + void testSequenceColumnLeftOutIsRejected() { + // the incoming ts would decide the group without ever being stored + assertThatThrownBy(() -> check(0, 1, 2)) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "The target write columns must cover the sequence group ordered by [ts] " + + "entirely or not at all, but [ts] is missing."); + } + + @Test + void testOnlySequenceColumnTargetedIsRejected() { + assertThatThrownBy(() -> check(0, 3)) + .isInstanceOf(InvalidTargetColumnException.class) + .hasMessage( + "The target write columns must cover the sequence group ordered by [ts] " + + "entirely or not at all, but [status, amount] are missing."); + } + + @Test + void testSchemaWithoutSequenceGroupAcceptsAnyTarget() { + assertThatCode( + () -> + TargetColumns.checkSequenceGroupsAreFullyTargeted( + TWO_COL_SCHEMA, new int[] {0})) + .doesNotThrowAnyException(); + } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java index bd1bd3217b8..3ead00ccc71 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java @@ -20,13 +20,17 @@ import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.config.TableConfig; +import org.apache.fluss.metadata.AggFunctionType; import org.apache.fluss.metadata.AggFunctions; import org.apache.fluss.metadata.DeleteBehavior; +import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.Schema; import org.apache.fluss.metadata.SchemaInfo; import org.apache.fluss.record.BinaryValue; import org.apache.fluss.record.TestingSchemaGetter; import org.apache.fluss.row.BinaryRow; +import org.apache.fluss.server.kv.rowmerger.aggregate.AggregationContext; +import org.apache.fluss.server.kv.rowmerger.aggregate.functions.FieldLastValueAgg; import org.apache.fluss.types.DataTypes; import org.apache.fluss.types.RowType; @@ -999,6 +1003,258 @@ void testPartialAggregateRowMergerDeleteAllScenarios() { } } + // --------------------------------------------------------------------------------------------- + // sequence groups + // + // With aggregate functions a sequence group acts as an ordering key rather than a version + // filter: a stale row still aggregates, only as one that happened earlier, while a row without + // any sequence for the group contributes nothing at all. + // --------------------------------------------------------------------------------------------- + + /** + * {@code total} accumulates under the order of {@code ts}, while {@code note} takes part in no + * group and keeps the plain last-value behavior. + */ + private static final Schema SCHEMA_SEQUENCE_GROUP = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("total", DataTypes.BIGINT(), AggFunctions.SUM()) + .column("ts", DataTypes.INT()) + .column("note", DataTypes.STRING()) + .sequenceGroup( + java.util.Collections.singletonList("ts"), + java.util.Collections.singletonList("total")) + .primaryKey("id") + .build(); + + private BinaryValue sequenceGroupRow(Long total, Integer ts, String note) { + return toBinaryValue( + compactedRow( + SCHEMA_SEQUENCE_GROUP.getRowType(), new Object[] {1, total, ts, note})); + } + + private AggregateRowMerger sequenceGroupMerger() { + TableConfig tableConfig = new TableConfig(new Configuration()); + AggregateRowMerger merger = createMerger(SCHEMA_SEQUENCE_GROUP, tableConfig); + merger.configureTargetColumns(null, SCHEMA_ID, SCHEMA_SEQUENCE_GROUP); + return merger; + } + + @Test + void testForwardAdvancesTheSequenceAndAggregates() { + AggregateRowMerger merger = sequenceGroupMerger(); + BinaryValue stored = sequenceGroupRow(30L, 100, "first"); + + BinaryValue merged = merger.merge(stored, sequenceGroupRow(20L, 200, "second")); + assertThat(merged.row.getLong(1)).isEqualTo(50L); // 30 + 20 + assertThat(merged.row.getInt(2)).isEqualTo(200); // the sequence moves forward + assertThat(merged.row.getString(3).toString()).isEqualTo("second"); + + // an equal sequence advances as well, so a replayed record still refreshes the group + BinaryValue replayed = merger.merge(stored, sequenceGroupRow(20L, 100, "same")); + assertThat(replayed.row.getLong(1)).isEqualTo(50L); + assertThat(replayed.row.getInt(2)).isEqualTo(100); + } + + @Test + void testStaleStillAggregatesButKeepsTheSequence() { + AggregateRowMerger merger = sequenceGroupMerger(); + BinaryValue stored = sequenceGroupRow(30L, 100, "first"); + + // the incoming row is older, yet its amount is a fact that belongs in the total + BinaryValue merged = merger.merge(stored, sequenceGroupRow(10L, 50, "older")); + assertThat(merged.row.getLong(1)).isEqualTo(40L); // 30 + 10 + assertThat(merged.row.getInt(2)).isEqualTo(100); // the sequence does not go backwards + } + + @Test + void testStaleRecordTakesItsEarlierPositionForAnOrderSensitiveFunction() { + // a stale record is aggregated into its earlier position, so the launch price wins over the + // later repricing + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column( + "first_price", + DataTypes.BIGINT(), + AggFunctions.of(AggFunctionType.FIRST_VALUE)) + .column("ts", DataTypes.INT()) + .sequenceGroup( + java.util.Collections.singletonList("ts"), + java.util.Collections.singletonList("first_price")) + .primaryKey("id") + .build(); + TableConfig tableConfig = new TableConfig(new Configuration()); + AggregateRowMerger merger = createMerger(schema, tableConfig); + merger.configureTargetColumns(null, SCHEMA_ID, schema); + + BinaryValue stored = + toBinaryValue(compactedRow(schema.getRowType(), new Object[] {1, 100L, 200})); + BinaryValue merged = + merger.merge( + stored, + toBinaryValue( + compactedRow(schema.getRowType(), new Object[] {1, 80L, 50}))); + + assertThat(merged.row.getLong(1)).isEqualTo(80L); // the launch price wins + assertThat(merged.row.getInt(2)).isEqualTo(200); // the sequence does not go backwards + } + + @Test + void testAllSkippedWriteReturnsTheStoredValueAsItIs() { + // every field belongs to the group, so an all-NULL sequence makes the whole write a no-op + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("total", DataTypes.BIGINT(), AggFunctions.SUM()) + .column("ts", DataTypes.INT()) + .sequenceGroup( + java.util.Collections.singletonList("ts"), + java.util.Collections.singletonList("total")) + .primaryKey("id") + .build(); + TableConfig tableConfig = new TableConfig(new Configuration()); + AggregateRowMerger merger = createMerger(schema, tableConfig); + merger.configureTargetColumns(null, SCHEMA_ID, schema); + + BinaryValue stored = + toBinaryValue(compactedRow(schema.getRowType(), new Object[] {1, 30L, 100})); + BinaryValue skipped = + toBinaryValue(compactedRow(schema.getRowType(), new Object[] {1, 5L, null})); + assertThat(merger.merge(stored, skipped)).isSameAs(stored); + } + + @Test + void testGroupWithoutAnySequenceContributesNothing() { + AggregateRowMerger merger = sequenceGroupMerger(); + BinaryValue stored = sequenceGroupRow(30L, 100, "first"); + + // no sequence at all, so the group is skipped and the amount is not accumulated + BinaryValue merged = merger.merge(stored, sequenceGroupRow(20L, null, "dropped")); + assertThat(merged.row.getLong(1)).isEqualTo(30L); + assertThat(merged.row.getInt(2)).isEqualTo(100); + // the column outside the group is unaffected by the skip + assertThat(merged.row.getString(3).toString()).isEqualTo("dropped"); + } + + @Test + void testFirstRowIsArbitrated() { + AggregateRowMerger merger = sequenceGroupMerger(); + + // a first row without a sequence contributes nothing to the group, while the ungrouped + // field is accepted + assertThat(merger.merge(null, sequenceGroupRow(30L, null, "outside"))) + .isEqualTo(sequenceGroupRow(null, null, "outside")); + + // a first row carrying a sequence is fully accepted without being re-encoded + BinaryValue first = sequenceGroupRow(30L, 100, "first"); + assertThat(merger.merge(null, first)).isSameAs(first); + } + + @Test + void testGroupsAreArbitratedIndependently() { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("paid", DataTypes.BIGINT(), AggFunctions.SUM()) + .column("pay_ts", DataTypes.INT()) + .column("shipped", DataTypes.BIGINT(), AggFunctions.SUM()) + .column("ship_ts", DataTypes.INT()) + .sequenceGroup( + java.util.Collections.singletonList("pay_ts"), + java.util.Collections.singletonList("paid")) + .sequenceGroup( + java.util.Collections.singletonList("ship_ts"), + java.util.Collections.singletonList("shipped")) + .primaryKey("id") + .build(); + TableConfig tableConfig = new TableConfig(new Configuration()); + AggregateRowMerger merger = createMerger(schema, tableConfig); + merger.configureTargetColumns(null, SCHEMA_ID, schema); + RowType rowType = schema.getRowType(); + + BinaryValue stored = + toBinaryValue(compactedRow(rowType, new Object[] {1, 30L, 100, 30L, 100})); + // the pay group moves forward while the ship group carries no sequence at all + BinaryValue merged = + merger.merge( + stored, + toBinaryValue( + compactedRow(rowType, new Object[] {1, 20L, 200, 20L, null}))); + + assertThat(merged.row.getLong(1)).isEqualTo(50L); // paid accumulated + assertThat(merged.row.getInt(2)).isEqualTo(200); // pay sequence advanced + assertThat(merged.row.getLong(3)).isEqualTo(30L); // shipped skipped entirely + assertThat(merged.row.getInt(4)).isEqualTo(100); // ship sequence unchanged + } + + @Test + void testOrderIndependentFunctionGivesTheSameTotalWhateverTheArrivalOrder() { + BinaryValue newer = sequenceGroupRow(20L, 200, "newer"); + BinaryValue older = sequenceGroupRow(10L, 50, "older"); + + // in order: the newer row lands second + AggregateRowMerger inOrder = sequenceGroupMerger(); + BinaryValue inOrderResult = inOrder.merge(sequenceGroupRow(30L, 100, "first"), newer); + inOrderResult = inOrder.merge(inOrderResult, older); + + // out of order: the older row lands second + AggregateRowMerger outOfOrder = sequenceGroupMerger(); + BinaryValue outOfOrderResult = outOfOrder.merge(sequenceGroupRow(30L, 100, "first"), older); + outOfOrderResult = outOfOrder.merge(outOfOrderResult, newer); + + // sum is order independent, so both arrive at the same total and the same sequence + assertThat(inOrderResult.row.getLong(1)).isEqualTo(60L); + assertThat(outOfOrderResult.row.getLong(1)).isEqualTo(60L); + assertThat(inOrderResult.row.getInt(2)).isEqualTo(200); + assertThat(outOfOrderResult.row.getInt(2)).isEqualTo(200); + } + + @Test + void testPartialUpdateArbitratesTheWrittenColumnsOnly() { + TableConfig tableConfig = new TableConfig(new Configuration()); + AggregateRowMerger merger = createMerger(SCHEMA_SEQUENCE_GROUP, tableConfig); + // 'note' is left out of the write, so it keeps the stored value whatever the group decides, + // where a full row write would have it take the incoming value + RowMerger partial = + merger.configureTargetColumns( + new int[] {0, 1, 2}, SCHEMA_ID, SCHEMA_SEQUENCE_GROUP); + + // the partial path applies the same first-write arbitration as the full aggregation path + assertThat(partial.merge(null, sequenceGroupRow(5L, null, null))) + .isEqualTo(sequenceGroupRow(null, null, null)); + + BinaryValue stored = sequenceGroupRow(30L, 100, "kept"); + + // the group moves forward, so the written columns aggregate and the sequence follows + BinaryValue forward = partial.merge(stored, sequenceGroupRow(20L, 200, null)); + assertThat(forward.row.getLong(1)).isEqualTo(50L); // 30 + 20 + assertThat(forward.row.getInt(2)).isEqualTo(200); + assertThat(forward.row.getString(3).toString()).isEqualTo("kept"); + + // a stale row still aggregates, only in reverse, and leaves the sequence where it was + BinaryValue stale = partial.merge(stored, sequenceGroupRow(10L, 50, null)); + assertThat(stale.row.getLong(1)).isEqualTo(40L); // 30 + 10 + assertThat(stale.row.getInt(2)).isEqualTo(100); + assertThat(stale.row.getString(3).toString()).isEqualTo("kept"); + + // no sequence at all and every written field is grouped, so the write is a no-op and the + // stored value is returned as is + assertThat(partial.merge(stored, sequenceGroupRow(5L, null, null))).isSameAs(stored); + } + + @Test + void testSequenceColumnIsNotAggregated() { + // a sequence column must not take an aggregate function of its own, otherwise a stale row + // could move the sequence backwards. the merger keeps it under the order of its own group, + // which the stale case above already asserts, and here it is checked on the aggregators. + AggregationContext context = + AggregationContext.create(SCHEMA_SEQUENCE_GROUP, KvFormat.COMPACTED); + assertThat(context.getSequenceGroups()).isNotNull(); + // index 2 is 'ts', which reports last_value rather than a sum or the default + assertThat(context.getAggregators()[2]).isInstanceOf(FieldLastValueAgg.class); + } + private AggregateRowMerger createMerger(Schema schema, TableConfig tableConfig) { TestingSchemaGetter schemaGetter = new TestingSchemaGetter(new SchemaInfo(schema, SCHEMA_ID)); diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMergerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMergerTest.java index 3f445d0a4c5..5e464b7ab9f 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMergerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/DefaultRowMergerTest.java @@ -23,6 +23,7 @@ import org.apache.fluss.record.BinaryValue; import org.apache.fluss.types.DataTypes; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; @@ -30,8 +31,9 @@ import static org.apache.fluss.testutils.DataTestUtils.compactedRow; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; -/** Tests for {@link DefaultRowMerger} delete behavior functionality. */ +/** Tests for {@link DefaultRowMerger} delete behavior and sequence group arbitration. */ class DefaultRowMergerTest { private static final Schema SCHEMA = @@ -114,4 +116,138 @@ void testPartialUpdateRowMergerDeleteBehavior(DeleteBehavior deleteBehavior) { assertThat(partialMerger.merge(oldValue, newValue)).isEqualTo(mergeValue); assertThat(partialMerger.delete(mergeValue)).isEqualTo(createBinaryValue(1, "old", null)); } + + /** {@code name} is ordered by {@code ts}, while {@code note} takes part in no group. */ + private static final Schema SEQUENCE_GROUP_SCHEMA = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("ts", DataTypes.INT()) + .column("note", DataTypes.STRING()) + .sequenceGroup( + java.util.Collections.singletonList("ts"), + java.util.Collections.singletonList("name")) + .primaryKey("id") + .build(); + + private static BinaryValue sequenceGroupValue(String name, Integer ts, String note) { + return new BinaryValue( + (short) 1, + compactedRow(SEQUENCE_GROUP_SCHEMA.getRowType(), new Object[] {1, name, ts, note})); + } + + @Test + void testSequenceGroupRowMergerOnFullRow() { + RowMerger merger = + new DefaultRowMerger(KvFormat.COMPACTED, DeleteBehavior.ALLOW) + .configureTargetColumns(null, (short) 1, SEQUENCE_GROUP_SCHEMA); + + // a first row without a sequence skips the group, while an ungrouped field is accepted + assertThat(merger.merge(null, sequenceGroupValue("skipped", null, "n0"))) + .isEqualTo(sequenceGroupValue(null, null, "n0")); + + // a first row carrying a sequence initializes the group without being re-encoded + BinaryValue first = sequenceGroupValue("first", 100, "n1"); + assertThat(merger.merge(null, first)).isSameAs(first); + + // every group advances, so the whole incoming row wins without being re-encoded + BinaryValue newer = sequenceGroupValue("newer", 101, "n2"); + assertThat(merger.merge(first, newer)).isSameAs(newer); + + // the group falls behind, so its columns keep the stored values while the column outside + // any group still takes the incoming one + BinaryValue stale = sequenceGroupValue("stale", 99, "n3"); + assertThat(merger.merge(newer, stale)).isEqualTo(sequenceGroupValue("newer", 101, "n3")); + + // a delete carries no sequence values, so it keeps removing the whole row + assertThat(merger.delete(newer)).isNull(); + } + + @Test + void testSequenceGroupRowMergerReadsAShorterStoredRow() { + RowMerger merger = + new DefaultRowMerger(KvFormat.COMPACTED, DeleteBehavior.ALLOW) + .configureTargetColumns(null, (short) 1, SEQUENCE_GROUP_SCHEMA); + + // the stored row was written before 'ts' and 'note' were added, so it carries fewer fields + // and its absent sequence orders before everything + BinaryValue shortRow = + new BinaryValue( + (short) 1, compactedRow(SCHEMA.getRowType(), new Object[] {1, "stored"})); + BinaryValue incoming = sequenceGroupValue("incoming", 1, "n1"); + assertThat(merger.merge(shortRow, incoming)).isSameAs(incoming); + + // without any sequence the incoming group is dropped, and the missing fields of the stored + // row are read as null + BinaryValue withoutSequence = sequenceGroupValue("dropped", null, "n2"); + assertThat(merger.merge(shortRow, withoutSequence)) + .isEqualTo(sequenceGroupValue("stored", null, "n2")); + } + + @Test + void testSequenceGroupRowMergerOnPartialColumns() { + DefaultRowMerger merger = new DefaultRowMerger(KvFormat.COMPACTED, DeleteBehavior.ALLOW); + // only 'name' and its sequence column are written, leaving 'note' out + RowMerger partialMerger = + merger.configureTargetColumns( + new int[] {0, 1, 2}, (short) 1, SEQUENCE_GROUP_SCHEMA); + + // the partial path applies the same first-write arbitration as the full-row path + assertThat(partialMerger.merge(null, sequenceGroupValue("skipped", null, null))) + .isEqualTo(sequenceGroupValue(null, null, null)); + + BinaryValue stored = sequenceGroupValue("stored", 100, "kept"); + // the group advances, so the written columns take the incoming values and 'note' is kept + assertThat(partialMerger.merge(stored, sequenceGroupValue("newer", 101, null))) + .isEqualTo(sequenceGroupValue("newer", 101, "kept")); + // the group falls behind, so the write is a no-op and the stored value is returned as is + assertThat(partialMerger.merge(stored, sequenceGroupValue("stale", 99, null))) + .isSameAs(stored); + } + + @Test + void testBlindOverwriteRestoresAnAlreadyDecidedValue() { + // recovering by undo writes back the value stored at the checkpoint, which is older than + // what is in the store, so arbitrating it would discard the very row being recovered + BinaryValue checkpointed = sequenceGroupValue("checkpointed", 100, "n1"); + BinaryValue newer = sequenceGroupValue("newer", 200, "n2"); + + DefaultRowMerger blind = DefaultRowMerger.forBlindOverwrite(KvFormat.COMPACTED); + assertThat(blind.configureTargetColumns(null, (short) 1, SEQUENCE_GROUP_SCHEMA)) + // staying a DefaultRowMerger also keeps the fast path KvTablet takes for a write + // that may skip reading the stored row + .isSameAs(blind); + assertThat(blind.merge(newer, checkpointed)).isSameAs(checkpointed); + + RowMerger blindPartial = + DefaultRowMerger.forBlindOverwrite(KvFormat.COMPACTED) + .configureTargetColumns( + new int[] {0, 1, 2}, (short) 1, SEQUENCE_GROUP_SCHEMA); + assertThat(blindPartial.merge(newer, checkpointed)) + .isEqualTo(sequenceGroupValue("checkpointed", 100, "n2")); + } + + @Test + void testArbitratingMergerReplacesThePlainOne() { + DefaultRowMerger merger = new DefaultRowMerger(KvFormat.COMPACTED, DeleteBehavior.ALLOW); + + // KvTablet skips reading the stored row while the merger is a DefaultRowMerger, which would + // leave every group unarbitrated, so a schema with sequence groups must replace it + assertThat(merger.configureTargetColumns(null, (short) 1, SEQUENCE_GROUP_SCHEMA)) + .isNotInstanceOf(DefaultRowMerger.class); + // a schema without sequence groups keeps the plain merger and so keeps the fast path + assertThat(merger.configureTargetColumns(null, (short) 2, SCHEMA)).isSameAs(merger); + // the merger is rebuilt on a schema change and reused within one schema + RowMerger arbitrating = + merger.configureTargetColumns(null, (short) 3, SEQUENCE_GROUP_SCHEMA); + assertThat(merger.configureTargetColumns(null, (short) 3, SEQUENCE_GROUP_SCHEMA)) + .isSameAs(arbitrating); + + assertThatThrownBy( + () -> + arbitrating.configureTargetColumns( + null, (short) 3, SEQUENCE_GROUP_SCHEMA)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("does not support reconfigure"); + } } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/SequenceGroupsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/SequenceGroupsTest.java new file mode 100644 index 00000000000..99c25423a3c --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/SequenceGroupsTest.java @@ -0,0 +1,278 @@ +/* + * 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.fluss.server.kv.rowmerger; + +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.row.InternalRow; +import org.apache.fluss.types.DataType; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import javax.annotation.Nullable; + +import java.util.BitSet; +import java.util.stream.Stream; + +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static org.apache.fluss.testutils.DataTestUtils.compactedRow; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for how {@link SequenceGroups} arbitrates the groups declared on a schema. */ +class SequenceGroupsTest { + + /** + * {@code a} is ordered by {@code g1} and {@code b} by {@code g2}, so the groups are disjoint. + */ + private static final Schema TWO_GROUPS = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .column("g1", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .column("g2", DataTypes.INT()) + .sequenceGroup(singletonList("g1"), singletonList("a")) + .sequenceGroup(singletonList("g2"), singletonList("b")) + .primaryKey("k") + .build(); + + private static final int A = 1; + private static final int G1 = 2; + private static final int B = 3; + private static final int G2 = 4; + + private static InternalRow twoGroupsRow(@Nullable Integer g1, @Nullable Integer g2) { + return compactedRow(TWO_GROUPS.getRowType(), new Object[] {1, "a", g1, "b", g2}); + } + + /** Arbitrates the two rows and collects whether every field is accepted. */ + private static boolean[] acceptanceOf( + SequenceGroups groups, @Nullable InternalRow oldRow, InternalRow newRow) { + groups.arbitrate(oldRow, newRow); + boolean[] acceptance = new boolean[groups.fieldCount()]; + for (int i = 0; i < acceptance.length; i++) { + acceptance[i] = groups.accepts(i); + } + return acceptance; + } + + @Test + void testEachGroupIsArbitratedOnItsOwn() { + SequenceGroups groups = SequenceGroups.create(TWO_GROUPS); + + // the first group moves forward while the second falls behind + boolean[] acceptance = acceptanceOf(groups, twoGroupsRow(100, 100), twoGroupsRow(101, 99)); + assertThat(acceptance[A]).isTrue(); + assertThat(acceptance[B]).isFalse(); + // a sequence column follows the group it orders, so that its value stays in step with the + // columns arbitrated by it + assertThat(acceptance[G1]).isTrue(); + assertThat(acceptance[G2]).isFalse(); + // the primary key takes part in no group and is never held back + assertThat(acceptance[0]).isTrue(); + } + + @Test + void testGroupAdvancesOnAnEqualSequenceButNotWithoutOne() { + SequenceGroups groups = SequenceGroups.create(TWO_GROUPS); + + // a replayed record still refreshes the group + assertThat(acceptanceOf(groups, twoGroupsRow(100, 100), twoGroupsRow(100, 100))) + .containsOnly(true); + + // the incoming group carries no order information at all, so its values are dropped even + // though there is no stored row to compare against + boolean[] acceptance = acceptanceOf(groups, null, twoGroupsRow(null, 1)); + assertThat(acceptance[A]).isFalse(); + assertThat(acceptance[B]).isTrue(); + } + + @Test + void testGroupResolutionIsIndependentOfTheDeclarationShape() { + // the sequence column is declared before the column it orders + Schema sequenceFirst = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("g", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .sequenceGroup(singletonList("g"), singletonList("a")) + .primaryKey("k") + .build(); + InternalRow storedFirst = + compactedRow(sequenceFirst.getRowType(), new Object[] {1, 100, "a"}); + InternalRow incomingFirst = + compactedRow(sequenceFirst.getRowType(), new Object[] {1, 99, "a"}); + // both the sequence column and the column it orders are held back together + assertThat(acceptanceOf(SequenceGroups.create(sequenceFirst), storedFirst, incomingFirst)) + .containsExactly(true, false, false); + + // two columns naming the same sequence column advance as one group + Schema shared = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .column("b", DataTypes.STRING()) + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), asList("a", "b")) + .primaryKey("k") + .build(); + InternalRow storedShared = + compactedRow(shared.getRowType(), new Object[] {1, "a", "b", 100}); + InternalRow incomingShared = + compactedRow(shared.getRowType(), new Object[] {1, "a", "b", 99}); + assertThat(acceptanceOf(SequenceGroups.create(shared), storedShared, incomingShared)) + .containsExactly(true, false, false, false); + } + + @Test + void testMissingSequenceColumnInAShorterRowIsTheOldest() { + // a row written under an older schema carries fewer fields, so the sequence column is + // absent + Schema olderSchema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .primaryKey("k") + .build(); + SequenceGroups groups = SequenceGroups.create(TWO_GROUPS); + + InternalRow shortRow = compactedRow(olderSchema.getRowType(), new Object[] {1, "a"}); + assertThat(acceptanceOf(groups, shortRow, twoGroupsRow(1, 1))).containsOnly(true); + } + + // --------------------------------------------------------------------------------------------- + // composite sequence keys + // --------------------------------------------------------------------------------------------- + + /** {@code a} is ordered by {@code g1} and {@code g2} together, compared in that order. */ + private static final Schema COMPOSITE = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .column("g1", DataTypes.INT()) + .column("g2", DataTypes.INT()) + .sequenceGroup(asList("g1", "g2"), singletonList("a")) + .primaryKey("k") + .build(); + + private static InternalRow compositeRow(@Nullable Integer g1, @Nullable Integer g2) { + return compactedRow(COMPOSITE.getRowType(), new Object[] {1, "a", g1, g2}); + } + + private static boolean compositeAdvances( + @Nullable Integer storedG1, + @Nullable Integer storedG2, + @Nullable Integer incomingG1, + @Nullable Integer incomingG2) { + return acceptanceOf( + SequenceGroups.create(COMPOSITE), + compositeRow(storedG1, storedG2), + compositeRow(incomingG1, incomingG2))[A]; + } + + @Test + void testCompositeKeyComparesInTheDeclaredOrder() { + // the leading column decides on its own, whatever the trailing one says + assertThat(compositeAdvances(5, 100, 6, 1)).isTrue(); + assertThat(compositeAdvances(5, 100, 4, 999)).isFalse(); + // the leading columns tie, so the next one decides + assertThat(compositeAdvances(5, 100, 5, 101)).isTrue(); + assertThat(compositeAdvances(5, 100, 5, 99)).isFalse(); + // every column ties, which still advances the group + assertThat(compositeAdvances(5, 100, 5, 100)).isTrue(); + } + + @Test + void testCompositeKeyTreatsNullAsTheOldest() { + assertThat(compositeAdvances(null, 100, 1, 1)).isTrue(); + assertThat(compositeAdvances(1, 1, null, 999)).isFalse(); + // null equals null, so the leading column decides nothing and the trailing one arbitrates + assertThat(compositeAdvances(null, 100, null, 101)).isTrue(); + assertThat(compositeAdvances(null, 100, null, 99)).isFalse(); + // the group is dropped only when the incoming row carries no order information at all + assertThat(compositeAdvances(5, 100, null, null)).isFalse(); + assertThat(compositeAdvances(null, null, null, 1)).isTrue(); + } + + // --------------------------------------------------------------------------------------------- + // sequence column types + // --------------------------------------------------------------------------------------------- + + private static Stream supportedSequenceTypes() { + return Stream.of( + new Object[] {DataTypes.INT(), 101, 100}, + new Object[] {DataTypes.BIGINT(), 101L, 100L}, + new Object[] { + DataTypes.TIMESTAMP(), + org.apache.fluss.row.TimestampNtz.fromMillis(101), + org.apache.fluss.row.TimestampNtz.fromMillis(100) + }, + new Object[] { + DataTypes.TIMESTAMP_LTZ(), + org.apache.fluss.row.TimestampLtz.fromEpochMillis(101), + org.apache.fluss.row.TimestampLtz.fromEpochMillis(100) + }); + } + + @ParameterizedTest + @MethodSource("supportedSequenceTypes") + void testSupportedSequenceColumnTypesOrderTheirGroup( + DataType sequenceType, Object newer, Object older) { + Schema schema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .column("g", sequenceType) + .sequenceGroup(singletonList("g"), singletonList("a")) + .primaryKey("k") + .build(); + SequenceGroups groups = SequenceGroups.create(schema); + + InternalRow stored = compactedRow(schema.getRowType(), new Object[] {1, "a", older}); + InternalRow newerRow = compactedRow(schema.getRowType(), new Object[] {1, "a", newer}); + InternalRow withoutSequence = + compactedRow(schema.getRowType(), new Object[] {1, "a", null}); + + assertThat(acceptanceOf(groups, stored, newerRow)[A]).isTrue(); + assertThat(acceptanceOf(groups, newerRow, stored)[A]).isFalse(); + assertThat(acceptanceOf(groups, stored, withoutSequence)[A]).isFalse(); + } + + @Test + void testRestrictToLeavesUncoveredGroupsOutOfTheArbitration() { + BitSet coveringFirstGroup = new BitSet(); + coveringFirstGroup.set(0); // k + coveringFirstGroup.set(A); + coveringFirstGroup.set(G1); + SequenceGroups restricted = + SequenceGroups.create(TWO_GROUPS).restrictTo(coveringFirstGroup); + + // both groups fall behind, yet only the covered one still holds its fields back + boolean[] acceptance = + acceptanceOf(restricted, twoGroupsRow(100, 100), twoGroupsRow(99, 99)); + assertThat(acceptance).containsExactly(true, false, false, true, true); + + // the uncovered group no longer arbitrates b or g2, so a null sequence cannot hold them + // back + assertThat(acceptanceOf(restricted, twoGroupsRow(100, 100), twoGroupsRow(101, null))) + .containsExactly(true, true, true, true, true); + } +} diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/SequenceGroupValidationTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/SequenceGroupValidationTest.java new file mode 100644 index 00000000000..abe3af87c55 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/SequenceGroupValidationTest.java @@ -0,0 +1,123 @@ +/* + * 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.fluss.server.utils; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.exception.InvalidConfigException; +import org.apache.fluss.metadata.MergeEngineType; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +import org.apache.fluss.types.DataTypes; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import static java.util.Collections.singletonList; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Table-level rejection tests for sequence groups. These rejections depend on the merge engine or + * the table type (log vs primary key), so they live in {@link TableDescriptorValidation}. The + * schema-level rejections (existence, types, cross-group relations) are covered by {@code + * SchemaSequenceGroupTest}. + */ +class SequenceGroupValidationTest { + + /** A primary-key schema whose {@code a} is ordered by {@code g}. */ + private static Schema orderedByG() { + return Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), singletonList("a")) + .primaryKey("k") + .build(); + } + + private static void validate(Schema schema) { + validate(schema, null); + } + + private static void validate(Schema schema, MergeEngineType mergeEngine) { + TableDescriptor.Builder builder = + TableDescriptor.builder() + .schema(schema) + .distributedBy(1) + .property(ConfigOptions.TABLE_REPLICATION_FACTOR, 1); + if (mergeEngine != null) { + builder.property(ConfigOptions.TABLE_MERGE_ENGINE, mergeEngine); + } + TableDescriptorValidation.validateTableDescriptor(builder.build(), 1024, null); + } + + @Test + void testSchemaWithoutSequenceGroupIsNotAffected() { + Schema schema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .primaryKey("k") + .build(); + + assertThatCode(() -> validate(schema)).doesNotThrowAnyException(); + // a merge engine is only rejected together with a sequence group + assertThatCode(() -> validate(schema, MergeEngineType.FIRST_ROW)) + .doesNotThrowAnyException(); + } + + @Test + void testLogTableIsRejected() { + // nothing consults the sequence groups when merging, as there is no merging at all. Since + // a log table has no primary key, we build the schema without one. + Schema schema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .column("g", DataTypes.INT()) + .sequenceGroup(singletonList("g"), singletonList("a")) + .build(); + + assertThatThrownBy(() -> validate(schema)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining("Sequence group is only supported in primary key table."); + } + + @ParameterizedTest + @EnumSource( + value = MergeEngineType.class, + names = {"AGGREGATION"}, + mode = EnumSource.Mode.EXCLUDE) + void testMergeEngineWithoutSequenceGroupSupportIsRejected(MergeEngineType mergeEngine) { + assertThatThrownBy(() -> validate(orderedByG(), mergeEngine)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining( + String.format( + "Sequence group is not supported for '%s' merge engine.", + mergeEngine)); + } + + @Test + void testAggregationMergeEngineIsAccepted() { + // the aggregation engine reads the groups as an ordering key rather than a version filter, + // so it takes part in the arbitration instead of rejecting it + assertThatCode(() -> validate(orderedByG(), MergeEngineType.AGGREGATION)) + .doesNotThrowAnyException(); + } +} diff --git a/website/docs/table-design/merge-engines/aggregation.md b/website/docs/table-design/merge-engines/aggregation.md index 95fda27cfca..bdf3e1d227f 100644 --- a/website/docs/table-design/merge-engines/aggregation.md +++ b/website/docs/table-design/merge-engines/aggregation.md @@ -1088,6 +1088,49 @@ TableDescriptor.builder() +## Sequence Group + +Sequence groups give each aggregate field its own order. Order-dependent functions such as `first_value`, `last_value` +and `listagg` follow the arrival order by default, so an out-of-order record changes the result; under a sequence group +the stored sequence decides which record counts as first or last. Order-independent functions (`sum`, `product`, +`max`, `min`, `bool_and`, `bool_or`, `rbm32`, `rbm64`) give the same result either way. + +The example tracks the earliest price a product was offered at, where the price stream may arrive out of order: + +```sql title="Flink SQL" +CREATE TABLE products ( + k INT, + first_price BIGINT, + ts INT, + PRIMARY KEY (k) NOT ENFORCED +) WITH ( + 'table.merge-engine' = 'aggregation', + 'fields.first_price.agg' = 'first_value', + 'fields.ts.sequence-group' = 'first_price' +); + +-- a later repricing arrives first +INSERT INTO products VALUES (1, 100, 200); +-- the original launch price arrives late: its ts is older, so the record is stale and +-- aggregates in its earlier position, making 80 the first value +INSERT INTO products VALUES (1, 80, 50); +SELECT * FROM products; +-- Output: ++---+-------------+-----+ +| k | first_price | ts | ++---+-------------+-----+ +| 1 | 80 | 200 | ++---+-------------+-----+ +``` + +Without the sequence group `first_value` only sees the arrival order, so it would keep `100` and +miss the actual launch price. Note the stored sequence stays at `200`: the stale record takes its +earlier position without moving the sequence backwards. + +A record whose sequence columns are all NULL carries no order information and contributes nothing at all. See +[Handling out-of-order updates with sequence groups](../table-types/pk-table.md#handling-out-of-order-updates-with-sequence-groups) +for the declaration syntax, NULL behavior, and validation rules. + ## Delete Behavior The aggregation merge engine provides limited support for delete operations. You can configure the behavior using the `'table.delete.behavior'` option: diff --git a/website/docs/table-design/merge-engines/default.md b/website/docs/table-design/merge-engines/default.md index d4bc4c8c657..54db0cc84f4 100644 --- a/website/docs/table-design/merge-engines/default.md +++ b/website/docs/table-design/merge-engines/default.md @@ -79,4 +79,11 @@ SELECT * FROM T; +----+-----+----+ | 3 | 3.0 | t3 | +----+-----+----+ -``` \ No newline at end of file +``` + +## Sequence Group + +Sequence groups give each column group its own order, so a stale partial update is rejected independently for the +affected group: its fields keep the stored values while other groups still move forward. See +[Handling out-of-order updates with sequence groups](../table-types/pk-table.md#handling-out-of-order-updates-with-sequence-groups) +for the declaration syntax, NULL behavior, and validation rules. \ No newline at end of file diff --git a/website/docs/table-design/table-types/pk-table.md b/website/docs/table-design/table-types/pk-table.md index 54e22d50208..c4b13b9aac6 100644 --- a/website/docs/table-design/table-types/pk-table.md +++ b/website/docs/table-design/table-types/pk-table.md @@ -74,6 +74,103 @@ follows: | 1 | 2.0 | t1 | | 2 | 3.0 | t2 | +### Handling out-of-order updates with sequence groups + +Partial update keeps the last written value of each column, whether or not it is actually the newest record. When +several writers update the same row — a payment stream and a shipping stream, say — an out-of-order write silently +overwrites values that are already newer. + +A **sequence group** puts one or more columns under the order of a *sequence column*, so that those columns only take +an incoming value when the sequence column is not older than the stored one. Every group is arbitrated on its own, so +within a single write one group may move forward while another does not. + +#### Define a sequence group + +A sequence group is declared with the `'fields..sequence-group'` property, whose value lists the +columns it protects: + +```sql title="Flink SQL" +CREATE TABLE orders ( + order_id BIGINT, + pay_status STRING, + pay_time BIGINT, + ship_status STRING, + ship_time BIGINT, + PRIMARY KEY (order_id) NOT ENFORCED +) WITH ( + 'fields.pay_time.sequence-group' = 'pay_status', + 'fields.ship_time.sequence-group' = 'ship_status' +); + +INSERT INTO orders VALUES (1, 'paid', 100, 'shipped', 100); + +-- pay_time moves forward while ship_time falls behind, +-- so only the payment columns take the incoming values +INSERT INTO orders VALUES (1, 'refunded', 200, 'lost', 99); +SELECT * FROM orders; +-- Output: ++----------+------------+----------+-------------+-----------+ +| order_id | pay_status | pay_time | ship_status | ship_time | ++----------+------------+----------+-------------+-----------+ +| 1 | refunded | 200 | shipped | 100 | ++----------+------------+----------+-------------+-----------+ + +-- the shipping group catches up on its own, leaving the payment columns untouched +INSERT INTO orders VALUES (1, 'stale', 2, 'delivered', 300); +SELECT * FROM orders; +-- Output: ++----------+------------+----------+-------------+-----------+ +| order_id | pay_status | pay_time | ship_status | ship_time | ++----------+------------+----------+-------------+-----------+ +| 1 | refunded | 200 | delivered | 300 | ++----------+------------+----------+-------------+-----------+ +``` + +#### Use multiple sequence columns + +Naming more than one sequence column declares a composite sequence key. The columns are compared in the declared +order, and the first one that differs decides: + +```sql title="Flink SQL" +CREATE TABLE T ( + k INT, + v STRING, + epoch INT, + ts BIGINT, + PRIMARY KEY (k) NOT ENFORCED +) WITH ('fields.epoch,ts.sequence-group' = 'v'); +``` + +#### Handle NULL sequence values + +- A group takes the incoming values when its sequence columns are **not older** than the stored ones. Equal sequences + advance, so a replayed record still refreshes the group. +- A group whose incoming sequence columns are **all NULL** carries no order information and is skipped. +- NULL orders before every value, so a stored NULL is the oldest sequence. +- A sequence column is arbitrated by the very group it orders, keeping its value in step with the columns it + protects. +- `DELETE` is not arbitrated. A delete record carries the primary key alone and holds no sequence values to compare, + so it removes the whole row. + +#### Supported merge engines + +Sequence groups are supported by the [Default Merge Engine](../merge-engines/default.md) and the +[Aggregation Merge Engine](../merge-engines/aggregation.md); the pages describe what a group means under each engine. +A full-row write is arbitrated the same way as a partial one. + +#### Limitations + +A table is rejected at creation when: + +- it is a Log Table, or it configures the `first_row` or `versioned` merge engine, since neither consults the sequence + groups while merging; +- a sequence column doesn't exist in the schema, or its type is not one of `INT`, `BIGINT`, `TIMESTAMP` and + `TIMESTAMP_LTZ`; +- a primary key column is put into a group or used as a sequence column, since it holds the same value in both rows + being merged; +- a sequence column is put into another group, since it reports the order of its own group; +- the same column is declared by more than one group. + ## Merge Engines The **Merge Engine** in Fluss is a core component designed to efficiently handle and consolidate data updates for Primary Key Tables.