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..51fe22baaff 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 @@ -142,6 +142,11 @@ 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 columns.stream().anyMatch(col -> col.getSequenceColumns().isPresent()); + } + /** Returns the primary key indexes, if any, otherwise returns an empty array. */ public int[] getPrimaryKeyIndexes() { final List columns = getColumnNames(); @@ -365,7 +370,8 @@ public Builder fromColumns(List inputColumns) { column.dataType, column.comment, newColumnId, - column.aggFunction)); + column.aggFunction, + column.sequenceColumns)); } } @@ -488,6 +494,30 @@ public Builder withComment(@Nullable String comment) { return this; } + /** + * Apply the sequence columns ordering the previous column, i.e. put the previous column + * into the sequence group ordered by the given columns. + * + *

Passing more than one 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 previous column + */ + public Builder withSequenceColumns(String... sequenceColumns) { + checkNotNull(sequenceColumns, "Sequence columns must not be null."); + checkArgument(sequenceColumns.length > 0, "Sequence columns must not be empty."); + if (columns.isEmpty()) { + throw new IllegalArgumentException( + "Method 'withSequenceColumns(...)' must be called after a column definition, " + + "but there is no preceding column defined."); + } + columns.set( + columns.size() - 1, + columns.get(columns.size() - 1) + .withSequenceColumns(Arrays.asList(sequenceColumns))); + 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 @@ -589,6 +619,7 @@ public static final class Column implements Serializable { private final DataType dataType; private final @Nullable String comment; private final @Nullable AggFunction aggFunction; + private final @Nullable List sequenceColumns; public Column(String columnName, DataType dataType) { this(columnName, dataType, null, UNKNOWN_COLUMN_ID, null); @@ -609,11 +640,25 @@ public Column( @Nullable String comment, int columnId, @Nullable AggFunction aggFunction) { + this(columnName, dataType, comment, columnId, aggFunction, null); + } + + public Column( + String columnName, + DataType dataType, + @Nullable String comment, + int columnId, + @Nullable AggFunction aggFunction, + @Nullable List sequenceColumns) { this.columnName = columnName; this.dataType = dataType; this.comment = comment; this.columnId = columnId; this.aggFunction = aggFunction; + this.sequenceColumns = + sequenceColumns == null + ? null + : Collections.unmodifiableList(new ArrayList<>(sequenceColumns)); } public String getName() { @@ -641,12 +686,30 @@ public Optional getAggFunction() { return Optional.ofNullable(aggFunction); } + /** + * Gets the sequence columns ordering this column: it only takes an incoming value when they + * are not older than the stored ones. More than one means a composite key, compared in the + * listed order until one differs. + * + * @return the sequence columns, or empty if the column is merged without order arbitration + */ + public Optional> getSequenceColumns() { + return Optional.ofNullable(sequenceColumns); + } + public Column withComment(String comment) { - return new Column(columnName, dataType, comment, columnId, aggFunction); + return new Column( + columnName, dataType, comment, columnId, aggFunction, sequenceColumns); } public Column withAggFunction(@Nullable AggFunction aggFunction) { - return new Column(columnName, dataType, comment, columnId, aggFunction); + return new Column( + columnName, dataType, comment, columnId, aggFunction, sequenceColumns); + } + + public Column withSequenceColumns(@Nullable List sequenceColumns) { + return new Column( + columnName, dataType, comment, columnId, aggFunction, sequenceColumns); } @Override @@ -676,12 +739,14 @@ public boolean equals(Object o) { && Objects.equals(dataType, that.dataType) && Objects.equals(comment, that.comment) && Objects.equals(columnId, that.columnId) - && Objects.equals(aggFunction, that.aggFunction); + && Objects.equals(aggFunction, that.aggFunction) + && Objects.equals(sequenceColumns, that.sequenceColumns); } @Override public int hashCode() { - return Objects.hash(columnName, dataType, comment, columnId, aggFunction); + return Objects.hash( + columnName, dataType, comment, columnId, aggFunction, sequenceColumns); } } @@ -820,7 +885,8 @@ private static List normalizeColumns( column.getDataType().copy(false), column.getComment().isPresent() ? column.getComment().get() : null, column.getColumnId(), - column.getAggFunction().orElse(null))); + column.getAggFunction().orElse(null), + column.sequenceColumns)); } else { newColumns.add(column); } diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/json/ColumnJsonSerde.java b/fluss-common/src/main/java/org/apache/fluss/utils/json/ColumnJsonSerde.java index cbddfaceada..598e65e4016 100644 --- a/fluss-common/src/main/java/org/apache/fluss/utils/json/ColumnJsonSerde.java +++ b/fluss-common/src/main/java/org/apache/fluss/utils/json/ColumnJsonSerde.java @@ -27,8 +27,10 @@ import org.apache.fluss.types.DataType; import java.io.IOException; +import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import static org.apache.fluss.metadata.Schema.Column.UNKNOWN_COLUMN_ID; @@ -46,6 +48,7 @@ public class ColumnJsonSerde static final String AGG_FUNCTION = "agg_function"; static final String AGG_FUNCTION_TYPE = "type"; static final String AGG_FUNCTION_PARAMS = "parameters"; + static final String SEQUENCE_COLUMNS = "sequence_columns"; @Override public void serialize(Schema.Column column, JsonGenerator generator) throws IOException { @@ -71,6 +74,13 @@ public void serialize(Schema.Column column, JsonGenerator generator) throws IOEx } generator.writeEndObject(); } + if (column.getSequenceColumns().isPresent()) { + generator.writeArrayFieldStart(SEQUENCE_COLUMNS); + for (String sequenceColumn : column.getSequenceColumns().get()) { + generator.writeString(sequenceColumn); + } + generator.writeEndArray(); + } generator.writeNumberField(ID, column.getColumnId()); generator.writeEndObject(); @@ -105,11 +115,20 @@ public Schema.Column deserialize(JsonNode node) { } } + List sequenceColumns = null; + if (node.hasNonNull(SEQUENCE_COLUMNS)) { + sequenceColumns = new ArrayList<>(); + for (JsonNode sequenceColumn : node.get(SEQUENCE_COLUMNS)) { + sequenceColumns.add(sequenceColumn.asText()); + } + } + return new Schema.Column( columnName, dataType, node.hasNonNull(COMMENT) ? node.get(COMMENT).asText() : null, node.has(ID) ? node.get(ID).asInt() : UNKNOWN_COLUMN_ID, - aggFunction); + aggFunction, + sequenceColumns); } } diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/json/ColumnJsonSerdeTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/json/ColumnJsonSerdeTest.java index eba8159fb57..513cdb82b4e 100644 --- a/fluss-common/src/test/java/org/apache/fluss/utils/json/ColumnJsonSerdeTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/utils/json/ColumnJsonSerdeTest.java @@ -27,6 +27,7 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Collections; import static org.assertj.core.api.Assertions.assertThat; @@ -38,7 +39,7 @@ protected ColumnJsonSerdeTest() { @Override protected Schema.Column[] createObjects() { - Schema.Column[] columns = new Schema.Column[5]; + Schema.Column[] columns = new Schema.Column[6]; columns[0] = new Schema.Column("a", DataTypes.STRING()); columns[1] = new Schema.Column("b", DataTypes.INT(), "hello b"); columns[2] = new Schema.Column("c", new IntType(false), "hello c"); @@ -53,6 +54,9 @@ protected Schema.Column[] createObjects() { DataTypes.FIELD("g", DataTypes.STRING(), 1))), "hello c", (short) 2); + columns[5] = + new Schema.Column("h", DataTypes.STRING(), null, (short) 3) + .withSequenceColumns(Collections.singletonList("ts")); return columns; } @@ -63,7 +67,8 @@ protected String[] expectedJsons() { "{\"name\":\"b\",\"data_type\":{\"type\":\"INTEGER\"},\"comment\":\"hello b\",\"id\":-1}", "{\"name\":\"c\",\"data_type\":{\"type\":\"INTEGER\",\"nullable\":false},\"comment\":\"hello c\",\"id\":-1}", "{\"name\":\"d\",\"data_type\":{\"type\":\"INTEGER\",\"nullable\":false},\"comment\":\"hello c\",\"id\":2}", - "{\"name\":\"e\",\"data_type\":{\"type\":\"ROW\",\"fields\":[{\"name\":\"f\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":-1},{\"name\":\"g\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":1}]},\"comment\":\"hello c\",\"id\":2}" + "{\"name\":\"e\",\"data_type\":{\"type\":\"ROW\",\"fields\":[{\"name\":\"f\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":-1},{\"name\":\"g\",\"field_type\":{\"type\":\"STRING\"},\"field_id\":1}]},\"comment\":\"hello c\",\"id\":2}", + "{\"name\":\"h\",\"data_type\":{\"type\":\"STRING\"},\"sequence_columns\":[\"ts\"],\"id\":3}" }; } 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..8cdd1aa968a 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 @@ -95,6 +95,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. */ @@ -217,13 +220,21 @@ public static TableDescriptor toFlussTable(ResolvedCatalogBaseTable catalogBa // Check if aggregation merge engine is enabled to optimize parsing boolean isAggregationEngine = isAggregationMergeEngine(flinkTableConf); + // Sequence groups apply to the primary key table without merge engine, so they are parsed + // regardless of the merge engine and rejected server side when unsupported + Map> sequenceColumnsOf = parseSequenceGroups(flinkTableConf); + // Build schema with physical columns resolvedSchema.getColumns().stream() .filter(Column::isPhysical) .forEachOrdered( column -> addColumnToSchema( - schemBuilder, column, flinkTableConf, isAggregationEngine)); + schemBuilder, + column, + flinkTableConf, + isAggregationEngine, + sequenceColumnsOf)); // Configure auto-increment columns based on the 'auto-increment.fields' option. if (flinkTableConf.containsKey(AUTO_INCREMENT_FIELDS.key())) { @@ -744,18 +755,100 @@ 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, keyed by the sequence columns and + * listing the columns they protect: + * + * 

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

The returned mapping is inverted, giving the sequence columns of each protected column, + * which is how {@link Schema.Column} stores the relation. + */ + private static Map> parseSequenceGroups(Configuration tableConf) { + Map> sequenceColumnsOf = 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 = sequenceColumnsOf.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)); + } + } + } + return sequenceColumnsOf; + } + + /** + * 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 * @param tableConf the table configuration * @param parseAggFunction whether to parse aggregation function from config + * @param sequenceColumnsOf the sequence columns ordering each protected column */ private static void addColumnToSchema( Schema.Builder schemaBuilder, Column column, Configuration tableConf, - boolean parseAggFunction) { + boolean parseAggFunction, + Map> sequenceColumnsOf) { String columnName = column.getName(); DataType flussDataType = toFlussType(column.getDataType()); @@ -774,6 +867,12 @@ private static void addColumnToSchema( // Add comment if present column.getComment().ifPresent(schemaBuilder::withComment); + + // Put the column into the sequence group ordering it, if any + List sequenceColumns = sequenceColumnsOf.get(columnName); + if (sequenceColumns != null) { + schemaBuilder.withSequenceColumns(sequenceColumns.toArray(new String[0])); + } } private static Map extractCustomProperties( 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..803ccbb6c64 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,158 @@ 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 is skipped entirely + tEnv.executeSql("insert into agg_seq_group values (1, 5, cast(null as int))").await(); + assertResultsIgnoreOrder(rowIter, Arrays.asList("-U[1, 60, 200]", "+U[1, 60, 200]"), 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..696dfd4b60f 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 @@ -359,6 +359,96 @@ 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 columnName) { + return schema.getColumns().stream() + .filter(column -> column.getName().equals(columnName)) + .findFirst() + .flatMap(org.apache.fluss.metadata.Schema.Column::getSequenceColumns) + .orElse(null); + } + + private static void assertSequenceGroupRejected(String key, String value, String message) { + assertThatThrownBy(() -> convertWithOptions(sequenceGroup(key, value))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(message); + } + + @Test + void testSequenceGroupIsInvertedOntoTheProtectedColumns() { + org.apache.fluss.metadata.Schema schema = + convertWithOptions(sequenceGroup("fields.g1.sequence-group", "a, b")); + + // the declaration is keyed by the sequence column, while the schema stores the relation on + // each protected column, and the names are trimmed on the way + assertThat(schema.hasSequenceGroup()).isTrue(); + 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"); + } + + @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 testOptionConversions() { ConfigOption flinkOption = FlinkConversions.toFlinkOption(ConfigOptions.TABLE_KV_FORMAT); 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/partialupdate/PartialUpdater.java b/fluss-server/src/main/java/org/apache/fluss/server/kv/partialupdate/PartialUpdater.java index a7ce4bac9c5..cf444ac5698 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,9 +44,25 @@ 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; + this.sequenceGroups = sequenceGroups; for (int targetColumn : targetColumns) { partialUpdateCols.set(targetColumn); } @@ -97,6 +114,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 +127,18 @@ public BinaryValue updateRow(@Nullable BinaryValue oldValue, BinaryValue partial return oldValue; } + boolean[] acceptance = + sequenceGroups == null + ? null + : sequenceGroups.resolveAcceptance( + oldValue == null ? null : oldValue.row, partialValue.row); + 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) && (acceptance == null || acceptance[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 +163,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..d0ac5a40e16 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 @@ -102,7 +102,13 @@ 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); + oldValue.row, + newValue.row, + oldContext, + newContext, + targetContext, + targetContext.getSequenceGroups(), + encoder); BinaryRow mergedRow = encoder.finishRow(); return new BinaryValue(targetSchemaId, mergedRow); @@ -291,6 +297,7 @@ public BinaryValue merge(@Nullable BinaryValue oldValue, BinaryValue newValue) { newContext, targetContext, targetColumnIdBitSet, + targetContext.getSequenceGroups(), 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..23efd113ae5 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,44 @@ public RowMerger configureTargetColumns( @Nullable int[] targetColumns, short latestShemaId, Schema latestSchema) { if (targetColumns == null || TargetColumns.specifiesAllSchemaFieldIndexes(latestSchema, targetColumns)) { - return this; + return fullRowMerger(latestShemaId, latestSchema); } else { // 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 +173,97 @@ 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) { + if (oldValue == null) { + return newValue; + } + + boolean[] acceptance = sequenceGroups.resolveAcceptance(oldValue.row, newValue.row); + if (acceptsEveryField(acceptance)) { + // Every group advances, so the whole incoming row wins + return newValue; + } + + rowEncoder.startNewRow(); + for (int i = 0; i < fieldGetters.length; i++) { + InternalRow source = acceptance[i] ? newValue.row : oldValue.row; + // the stored row may follow an older schema with fewer fields, in which case the + // missing fields are null + if (source.getFieldCount() < i + 1) { + rowEncoder.encodeField(i, null); + } else { + rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(source)); + } + } + return new BinaryValue(targetSchemaId, rowEncoder.finishRow()); + } + + private static boolean acceptsEveryField(boolean[] acceptance) { + for (boolean accepted : acceptance) { + if (!accepted) { + return false; + } + } + return true; + } + + @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..641c503b712 --- /dev/null +++ b/fluss-server/src/main/java/org/apache/fluss/server/kv/rowmerger/SequenceGroups.java @@ -0,0 +1,312 @@ +/* + * 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.ArrayList; +import java.util.Arrays; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +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.Column#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. + * + *

Instances are immutable and hold no per-record state, so one instance serves all keys of a + * table. + */ +@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 readers of its sequence columns, in the declared comparison order. */ + private final SequenceReader[][] readersOfGroup; + + private SequenceGroups(int[] groupOfField, SequenceReader[][] readersOfGroup) { + this.groupOfField = groupOfField; + this.readersOfGroup = readersOfGroup; + } + + /** + * 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) { + if (!schema.hasSequenceGroup()) { + return null; + } + + RowType rowType = schema.getRowType(); + List columns = schema.getColumns(); + int fieldCount = columns.size(); + + // columns naming the very same sequence columns belong to one group, keyed by those names + // so that the group ids stay stable across equal schemas + Map, Integer> groupIds = new LinkedHashMap<>(); + List> sequenceColumnsOfGroup = new ArrayList<>(); + + int[] groupOfField = new int[fieldCount]; + Arrays.fill(groupOfField, NO_GROUP); + + for (int i = 0; i < fieldCount; i++) { + List sequenceColumns = columns.get(i).getSequenceColumns().orElse(null); + if (sequenceColumns == null) { + continue; + } + Integer groupId = groupIds.get(sequenceColumns); + if (groupId == null) { + groupId = sequenceColumnsOfGroup.size(); + groupIds.put(sequenceColumns, groupId); + sequenceColumnsOfGroup.add(sequenceColumns); + } + groupOfField[i] = groupId; + } + + SequenceReader[][] readersOfGroup = new SequenceReader[sequenceColumnsOfGroup.size()][]; + for (int groupId = 0; groupId < sequenceColumnsOfGroup.size(); groupId++) { + List sequenceColumns = sequenceColumnsOfGroup.get(groupId); + SequenceReader[] readers = new SequenceReader[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); + readers[i] = + createReader( + 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; + } + readersOfGroup[groupId] = readers; + } + + return new SequenceGroups(groupOfField, readersOfGroup); + } + + /** + * Resolves, for every field, whether it may take the value carried by the incoming row. + * + *

A field is held back only when the group arbitrating it doesn't advance. Fields taking + * part in no group keep their original behavior and always accept the incoming value. + * + * @param oldRow the stored row, or null when there is no stored row yet + * @param newRow the incoming row + */ + public boolean[] resolveAcceptance(@Nullable InternalRow oldRow, InternalRow newRow) { + Decision[] decisions = decideGroups(oldRow, newRow); + + boolean[] acceptance = new boolean[groupOfField.length]; + for (int i = 0; i < groupOfField.length; i++) { + // without aggregate functions a skipped group and a stale one both keep the stored + // values, so the two need no telling apart here + acceptance[i] = + groupOfField[i] == NO_GROUP || decisions[groupOfField[i]] == Decision.FORWARD; + } + return acceptance; + } + + /** + * Resolves, for every field, what the group arbitrating it makes of the incoming row. Fields + * taking part in no group always report {@link Decision#FORWARD}, keeping their original + * behavior. + * + *

Callers that aggregate need this rather than {@link #resolveAcceptance}, so that they can + * aggregate a stale row in reverse instead of dropping it. + * + * @param oldRow the stored row, or null when there is no stored row yet + * @param newRow the incoming row + */ + public Decision[] resolveDecisions(@Nullable InternalRow oldRow, InternalRow newRow) { + Decision[] decisions = decideGroups(oldRow, newRow); + + Decision[] ofField = new Decision[groupOfField.length]; + for (int i = 0; i < groupOfField.length; i++) { + ofField[i] = + groupOfField[i] == NO_GROUP ? Decision.FORWARD : decisions[groupOfField[i]]; + } + return ofField; + } + + /** Decides every group of the schema, indexed by group id. */ + private Decision[] decideGroups(@Nullable InternalRow oldRow, InternalRow newRow) { + Decision[] decisions = new Decision[readersOfGroup.length]; + for (int groupId = 0; groupId < readersOfGroup.length; groupId++) { + decisions[groupId] = decide(readersOfGroup[groupId], oldRow, newRow); + } + return decisions; + } + + /** + * Decides one group, by comparing its sequence columns in the declared order until one of them + * differs. + */ + private static Decision decide( + SequenceReader[] readers, @Nullable InternalRow oldRow, InternalRow newRow) { + Comparable[] newSequence = new Comparable[readers.length]; + boolean allNull = true; + for (int i = 0; i < readers.length; i++) { + newSequence[i] = readers[i].read(newRow); + if (newSequence[i] != null) { + allNull = false; + } + } + if (allNull) { + // the group carries no order information at all + return Decision.SKIP; + } + if (oldRow == null) { + return Decision.FORWARD; + } + + for (int i = 0; i < readers.length; i++) { + int comparison = compare(newSequence[i], readers[i].read(oldRow)); + 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; + } + + /** Null is treated as the smallest value, consistently with the versioned merge engine. */ + @SuppressWarnings("unchecked") + private static int compare(@Nullable Comparable left, @Nullable Comparable right) { + if (left == null) { + return right == null ? 0 : -1; + } + if (right == null) { + return 1; + } + return ((Comparable) left).compareTo(right); + } + + /** + * Returns a reader 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 SequenceReader createReader( + String columnName, DataType dataType, int fieldIndex) { + switch (dataType.getTypeRoot()) { + case INTEGER: + return row -> absent(row, fieldIndex) ? null : row.getInt(fieldIndex); + case BIGINT: + return row -> absent(row, fieldIndex) ? null : row.getLong(fieldIndex); + case TIMESTAMP_WITHOUT_TIME_ZONE: + int ntzPrecision = ((TimestampType) dataType).getPrecision(); + return row -> + absent(row, fieldIndex) + ? null + : row.getTimestampNtz(fieldIndex, ntzPrecision); + case TIMESTAMP_WITH_LOCAL_TIME_ZONE: + int ltzPrecision = ((LocalZonedTimestampType) dataType).getPrecision(); + return row -> + absent(row, fieldIndex) + ? null + : row.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); + } + + /** Reads the sequence value of a sequence column out of a row. */ + @FunctionalInterface + private interface SequenceReader extends Serializable { + + /** Returns the sequence value, or null if the column is absent or SQL NULL. */ + @Nullable + Comparable read(InternalRow row); + } +} 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..cf18170626a 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; @@ -59,6 +62,8 @@ private AggregateFieldsProcessor() {} * @param oldContext context for the old row schema * @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( @@ -67,10 +72,15 @@ public static void aggregateAllFieldsWithTargetSchema( 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 + SequenceGroups.Decision[] decisions = + sequenceGroups == null ? null : sequenceGroups.resolveDecisions(oldRow, newRow); + // Fast path: all three schemas are the same if (targetContext == oldContext && targetContext == newInputContext) { - aggregateAllFieldsWithSameSchema(oldRow, newRow, targetContext, encoder); + aggregateAllFieldsWithSameSchema(oldRow, newRow, targetContext, decisions, encoder); return; } @@ -101,11 +111,21 @@ public static void aggregateAllFieldsWithTargetSchema( oldRow, newRow, targetAggregators[targetIdx], + decisionOf(decisions, 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.Decision[] decisions, int fieldIndex) { + return decisions == null ? SequenceGroups.Decision.FORWARD : decisions[fieldIndex]; + } + /** * Aggregate and encode a single field. * @@ -114,6 +134,8 @@ public static void aggregateAllFieldsWithTargetSchema( * @param oldRow the old row * @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 */ @@ -123,11 +145,22 @@ private static void aggregateAndEncode( BinaryRow oldRow, BinaryRow newRow, FieldAggregator aggregator, + SequenceGroups.Decision decision, int targetIdx, RowEncoder encoder) { Object accumulator = 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); } @@ -165,6 +198,8 @@ private static void copyOldValueAndEncode( * @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( @@ -174,11 +209,15 @@ public static void aggregateTargetFieldsWithTargetSchema( AggregationContext newInputContext, AggregationContext targetContext, BitSet targetColumnIdBitSet, + @Nullable SequenceGroups sequenceGroups, RowEncoder encoder) { + SequenceGroups.Decision[] decisions = + sequenceGroups == null ? null : sequenceGroups.resolveDecisions(oldRow, newRow); + // Fast path: all three schemas are the same if (targetContext == oldContext && targetContext == newInputContext) { aggregateTargetFieldsWithSameSchema( - oldRow, newRow, targetContext, targetColumnIdBitSet, encoder); + oldRow, newRow, targetContext, targetColumnIdBitSet, decisions, encoder); return; } @@ -208,6 +247,7 @@ public static void aggregateTargetFieldsWithTargetSchema( oldRow, newRow, targetAggregators[targetIdx], + decisionOf(decisions, targetIdx), targetIdx, encoder); } else if (oldIdx != null) { @@ -229,7 +269,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.Decision[] decisions, + RowEncoder encoder) { InternalRow.FieldGetter[] fieldGetters = context.getFieldGetters(); FieldAggregator[] aggregators = context.getAggregators(); int fieldCount = context.getFieldCount(); @@ -241,6 +285,7 @@ private static void aggregateAllFieldsWithSameSchema( oldRow, newRow, aggregators[idx], + decisionOf(decisions, idx), idx, encoder); } @@ -256,6 +301,7 @@ private static void aggregateTargetFieldsWithSameSchema( BinaryRow newRow, AggregationContext context, BitSet targetColumnIdBitSet, + @Nullable SequenceGroups.Decision[] decisions, RowEncoder encoder) { InternalRow.FieldGetter[] fieldGetters = context.getFieldGetters(); FieldAggregator[] aggregators = context.getAggregators(); @@ -273,6 +319,7 @@ private static void aggregateTargetFieldsWithSameSchema( oldRow, newRow, aggregators[idx], + decisionOf(decisions, 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..eae7edf02aa 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.Column column : schema.getColumns()) { + column.getSequenceColumns().ifPresent(names::addAll); + } + return names; + } } 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..3af861d0278 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 @@ -51,6 +51,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.EnumSet; +import java.util.HashSet; import java.util.LinkedHashSet; import java.util.List; import java.util.Optional; @@ -402,6 +403,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 +458,103 @@ 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.hasSequenceGroup()) { + return; + } + + // both checks reject a configuration that would otherwise be silently ignored, as only the + // primary key table without merge engine, or with the aggregation one, consults the + // sequence groups when merging + 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)); + } + + RowType rowType = schema.getRowType(); + List primaryKeyNames = schema.getPrimaryKeyColumnNames(); + Set protectedColumnNames = new HashSet<>(); + for (Schema.Column column : schema.getColumns()) { + if (column.getSequenceColumns().isPresent()) { + protectedColumnNames.add(column.getName()); + } + } + EnumSet supportedTypes = + EnumSet.of( + DataTypeRoot.INTEGER, + DataTypeRoot.BIGINT, + DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE, + DataTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE); + + for (Schema.Column column : schema.getColumns()) { + List sequenceColumns = column.getSequenceColumns().orElse(null); + if (sequenceColumns == null) { + continue; + } + // a primary key column holds the same value in both rows being merged, so it can + // neither order a group nor be held back by one + if (primaryKeyNames.contains(column.getName())) { + throw new InvalidConfigException( + String.format( + "The primary key column '%s' must not be put in a sequence group.", + column.getName())); + } + for (String sequenceColumn : sequenceColumns) { + int columnIndex = rowType.getFieldIndex(sequenceColumn); + if (columnIndex < 0) { + throw new InvalidConfigException( + String.format( + "The sequence column '%s' doesn't exist in schema.", + sequenceColumn)); + } + if (primaryKeyNames.contains(sequenceColumn)) { + throw new InvalidConfigException( + String.format( + "The sequence column '%s' must not be a primary key column.", + sequenceColumn)); + } + if (protectedColumnNames.contains(sequenceColumn)) { + throw new InvalidConfigException( + String.format( + "The sequence column '%s' orders a sequence group, " + + "so it must not be put into another one.", + 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 (schema.getAggFunction(sequenceColumn).isPresent()) { + throw new InvalidConfigException( + 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 InvalidConfigException( + String.format( + "The sequence column '%s' must be one type of " + + "[INT, BIGINT, TIMESTAMP, TIMESTAMP_LTZ], but got %s.", + sequenceColumn, columnType)); + } + } + } + } + /** 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/rowmerger/AggregateRowMergerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/AggregateRowMergerTest.java index bd1bd3217b8..182d6b6bd84 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 @@ -22,11 +22,14 @@ import org.apache.fluss.config.TableConfig; 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 +1002,186 @@ 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()) + .withSequenceColumns("ts") + .column("ts", DataTypes.INT()) + .column("note", DataTypes.STRING()) + .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 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 testFirstRowIsTakenAsItIs() { + AggregateRowMerger merger = sequenceGroupMerger(); + + 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()) + .withSequenceColumns("pay_ts") + .column("pay_ts", DataTypes.INT()) + .column("shipped", DataTypes.BIGINT(), AggFunctions.SUM()) + .withSequenceColumns("ship_ts") + .column("ship_ts", DataTypes.INT()) + .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); + 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, so the group contributes nothing + BinaryValue skipped = partial.merge(stored, sequenceGroupRow(5L, null, null)); + assertThat(skipped.row.getLong(1)).isEqualTo(30L); + assertThat(skipped.row.getInt(2)).isEqualTo(100); + assertThat(skipped.row.getString(3).toString()).isEqualTo("kept"); + } + + @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..50a8dee628c 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,128 @@ 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()) + .withSequenceColumns("ts") + .column("ts", DataTypes.INT()) + .column("note", DataTypes.STRING()) + .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); + + // the first row of a key initializes every group + 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); + + 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 even the written columns keep the stored values + assertThat(partialMerger.merge(stored, sequenceGroupValue("stale", 99, null))) + .isEqualTo(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..468e89a8580 --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/kv/rowmerger/SequenceGroupsTest.java @@ -0,0 +1,275 @@ +/* + * 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.stream.Stream; + +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 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()) + .withSequenceColumns("g1") + .column("g1", DataTypes.INT()) + .column("b", DataTypes.STRING()) + .withSequenceColumns("g2") + .column("g2", DataTypes.INT()) + .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}); + } + + @Test + void testEachGroupIsArbitratedOnItsOwn() { + SequenceGroups groups = SequenceGroups.create(TWO_GROUPS); + + // the first group moves forward while the second falls behind + boolean[] acceptance = + groups.resolveAcceptance(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(groups.resolveAcceptance(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 = groups.resolveAcceptance(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()) + .withSequenceColumns("g") + .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( + SequenceGroups.create(sequenceFirst) + .resolveAcceptance(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()) + .withSequenceColumns("g") + .column("b", DataTypes.STRING()) + .withSequenceColumns("g") + .column("g", DataTypes.INT()) + .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(SequenceGroups.create(shared).resolveAcceptance(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(groups.resolveAcceptance(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()) + .withSequenceColumns("g1", "g2") + .column("g1", DataTypes.INT()) + .column("g2", DataTypes.INT()) + .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 SequenceGroups.create(COMPOSITE) + .resolveAcceptance( + 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()) + .withSequenceColumns("g") + .column("g", sequenceType) + .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(groups.resolveAcceptance(stored, newerRow)[A]).isTrue(); + assertThat(groups.resolveAcceptance(newerRow, stored)[A]).isFalse(); + assertThat(groups.resolveAcceptance(stored, withoutSequence)[A]).isFalse(); + } + + @Test + void testInvalidSequenceColumnIsRejectedWhenResolving() { + // table creation rejects these already, so resolving is only a backstop + Schema unsupportedType = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .withSequenceColumns("g") + .column("g", DataTypes.STRING()) + .primaryKey("k") + .build(); + assertThatThrownBy(() -> SequenceGroups.create(unsupportedType)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("must be one type of"); + + Schema missingColumn = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .column("a", DataTypes.STRING()) + .withSequenceColumns("missing") + .primaryKey("k") + .build(); + assertThatThrownBy(() -> SequenceGroups.create(missingColumn)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("doesn't exist in schema"); + } +} 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..b8a2c26a31d --- /dev/null +++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/SequenceGroupValidationTest.java @@ -0,0 +1,220 @@ +/* + * 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.AggFunctionType; +import org.apache.fluss.metadata.AggFunctions; +import org.apache.fluss.metadata.MergeEngineType; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.TableDescriptor; +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.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; + +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for the sequence group part of {@link TableDescriptorValidation}, which rejects at table + * creation what would otherwise be silently ignored or fail while merging. + */ +class SequenceGroupValidationTest { + + private static Schema.Builder pkSchema() { + return Schema.newBuilder().column("k", DataTypes.INT()).column("a", DataTypes.STRING()); + } + + /** A schema whose {@code a} is ordered by a {@code g} column of the given type. */ + private static Schema orderedByG(DataType sequenceType) { + return pkSchema() + .withSequenceColumns("g") + .column("g", sequenceType) + .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); + } + + private static Stream supportedSequenceTypes() { + return Stream.of( + DataTypes.INT(), + DataTypes.BIGINT(), + DataTypes.TIMESTAMP(), + DataTypes.TIMESTAMP_LTZ()); + } + + @ParameterizedTest + @MethodSource("supportedSequenceTypes") + void testSupportedSequenceColumnTypeIsAccepted(DataType sequenceType) { + assertThatCode(() -> validate(orderedByG(sequenceType))).doesNotThrowAnyException(); + } + + @Test + void testSchemaWithoutSequenceGroupIsNotAffected() { + Schema schema = pkSchema().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 testEveryColumnOfACompositeSequenceKeyIsChecked() { + Schema schema = + pkSchema() + .withSequenceColumns("g1", "g2") + .column("g1", DataTypes.INT()) + // only the trailing column has an unsupported type + .column("g2", DataTypes.STRING()) + .primaryKey("k") + .build(); + + assertThatThrownBy(() -> validate(schema)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining( + "The sequence column 'g2' must be one type of " + + "[INT, BIGINT, TIMESTAMP, TIMESTAMP_LTZ], but got STRING"); + } + + @Test + void testUnknownSequenceColumnIsRejected() { + Schema schema = pkSchema().withSequenceColumns("missing").primaryKey("k").build(); + + assertThatThrownBy(() -> validate(schema)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining("The sequence column 'missing' doesn't exist in schema."); + } + + @Test + void testLogTableIsRejected() { + // nothing consults the sequence groups when merging, as there is no merging at all + Schema schema = pkSchema().withSequenceColumns("g").column("g", DataTypes.INT()).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(DataTypes.INT()), 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(DataTypes.INT()), MergeEngineType.AGGREGATION)) + .doesNotThrowAnyException(); + } + + @Test + void testSequenceColumnWithAggregateFunctionIsRejected() { + Schema schema = + pkSchema() + .withSequenceColumns("g") + .column("g", DataTypes.INT(), AggFunctions.of(AggFunctionType.SUM)) + .primaryKey("k") + .build(); + + assertThatThrownBy(() -> validate(schema, MergeEngineType.AGGREGATION)) + .isInstanceOf(InvalidConfigException.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 + Schema schema = + Schema.newBuilder() + .column("k", DataTypes.INT()) + .withSequenceColumns("g") + .column("g", DataTypes.INT()) + .primaryKey("k") + .build(); + + assertThatThrownBy(() -> validate(schema)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining( + "The primary key column 'k' must not be put in a sequence group."); + } + + @Test + void testPrimaryKeyAsSequenceColumnIsRejected() { + Schema schema = pkSchema().withSequenceColumns("k").primaryKey("k").build(); + + assertThatThrownBy(() -> validate(schema)) + .isInstanceOf(InvalidConfigException.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 + Schema schema = + pkSchema() + .withSequenceColumns("pay_time") + .column("pay_time", DataTypes.TIMESTAMP()) + .withSequenceColumns("ship_time") + .column("ship_time", DataTypes.TIMESTAMP()) + .primaryKey("k") + .build(); + + assertThatThrownBy(() -> validate(schema)) + .isInstanceOf(InvalidConfigException.class) + .hasMessageContaining( + "The sequence column 'pay_time' orders a sequence group, " + + "so it must not be put into another one."); + } +} diff --git a/website/docs/table-design/merge-engines/aggregation.md b/website/docs/table-design/merge-engines/aggregation.md index 95fda27cfca..df05a478603 100644 --- a/website/docs/table-design/merge-engines/aggregation.md +++ b/website/docs/table-design/merge-engines/aggregation.md @@ -1088,6 +1088,106 @@ TableDescriptor.builder() +## Sequence Group + +Aggregate functions such as `sum` give the same result whatever order the records arrive in, but +`first_value`, `last_value` and `listagg` do not: they depend on which record is considered first or +last. Out of order records therefore produce a result that follows the arrival order rather than the +business order. + +A **sequence group** puts one or more columns under the order of a *sequence column*, giving the +engine an explicit order to follow. It is declared with the +`'fields..sequence-group'` property, whose value lists the columns it protects: + +```sql title="Flink SQL" +CREATE TABLE orders ( + k INT, + total BIGINT, + ts INT, + PRIMARY KEY (k) NOT ENFORCED +) WITH ( + 'table.merge-engine' = 'aggregation', + 'fields.total.agg' = 'sum', + 'fields.ts.sequence-group' = 'total' +); + +INSERT INTO orders VALUES (1, 30, 100); +-- the sequence moves forward, so the total accumulates and the sequence follows +INSERT INTO orders VALUES (1, 20, 200); +SELECT * FROM orders; +-- Output: ++---+-------+-----+ +| k | total | ts | ++---+-------+-----+ +| 1 | 50 | 200 | ++---+-------+-----+ + +-- an older record still accumulates, but leaves the stored sequence at 200 +INSERT INTO orders VALUES (1, 10, 50); +SELECT * FROM orders; +-- Output: ++---+-------+-----+ +| k | total | ts | ++---+-------+-----+ +| 1 | 60 | 200 | ++---+-------+-----+ +``` + +Each group is arbitrated on its own, so within a single write one group may move forward while +another does not. + +### Ordering key, not a version filter + +The meaning of a sequence group differs between this engine and the +[Default Merge Engine](table-design/merge-engines/default.md): + +| Incoming record | Default merge engine | Aggregation merge engine | +| ---------------------------------- | ------------------------ | ------------------------------------------------- | +| sequence not older than the stored | takes the incoming value | aggregates, and the sequence moves forward | +| sequence older than the stored | keeps the stored value | still aggregates, but the sequence stays put | +| no sequence at all (all NULL) | keeps the stored value | contributes nothing at all | + +Without aggregate functions a group acts as a version filter, dropping whatever is older. With them +it acts as an ordering key instead: an older record is a fact that still belongs in the total, so it +is aggregated as one that happened earlier. For order-independent functions (`sum`, `product`, +`max`, `min`, `bool_and`, `bool_or`, `rbm32`, `rbm64`) the order makes no difference to the result; +for the order-dependent ones the sequence decides which record counts as first or last. + +A record whose sequence columns are all NULL carries no order information at all and is skipped, so +its values are not aggregated. + +### Composite sequence key + +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, + total BIGINT, + epoch INT, + ts BIGINT, + PRIMARY KEY (k) NOT ENFORCED +) WITH ( + 'table.merge-engine' = 'aggregation', + 'fields.total.agg' = 'sum', + 'fields.epoch,ts.sequence-group' = 'total' +); +``` + +### Restrictions + +A table is rejected at creation when: + +- a sequence column has an aggregate function of its own, since the group it orders decides when it + advances and aggregating it would let a stale record move the sequence backwards; +- 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. + ## 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..6224b427f4c 100644 --- a/website/docs/table-design/merge-engines/default.md +++ b/website/docs/table-design/merge-engines/default.md @@ -79,4 +79,96 @@ SELECT * FROM T; +----+-----+----+ | 3 | 3.0 | t3 | +----+-----+----+ -``` \ No newline at end of file +``` + +## Sequence Group + +By default the latest write wins, whether or not it is actually the newest record. When several writers update the +same row, 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. This is what distinguishes a sequence group +from the [Versioned Merge Engine](table-design/merge-engines/versioned.md), which arbitrates the whole row with a +single version column. + +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 | ++----------+------------+----------+-------------+-----------+ +``` + +Sequence groups apply to a full-row write as well as to a [Partial Update](table-design/table-types/pk-table.md#partial-update). + +### Composite sequence key + +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'); +``` + +### Semantics + +- 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. + +### Restrictions + +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. The [Aggregation Merge Engine](table-design/merge-engines/aggregation.md) does support them, + where a group acts as an ordering key rather than a version filter; +- 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. \ No newline at end of file