Copilot commented on code in PR #1984:
URL: https://github.com/apache/fluss/pull/1984#discussion_r2559281612
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -179,11 +181,53 @@ public void addSplitsBack(List<TieringSplit> splits, int
subtaskId) {
@Override
public void addReader(int subtaskId) {
LOG.info("Adding reader: {} to Tiering Source enumerator.", subtaskId);
- if (context.registeredReaders().containsKey(subtaskId)) {
+ Map<Integer, ReaderInfo> readerByAttempt =
+ context.registeredReadersOfAttempts().get(subtaskId);
Review Comment:
The production code calls `context.registeredReadersOfAttempts()`, but the
standard Flink `SplitEnumeratorContext` interface doesn't have this method.
This method only exists in the test mock `FlussMockSplitEnumeratorContext`. The
code will fail at runtime in production with a compilation error or
`NoSuchMethodError` since the Flink-provided `SplitEnumeratorContext` doesn't
define `registeredReadersOfAttempts()`.
To fix this, you need to either:
1. Create a custom wrapper/extension of `SplitEnumeratorContext` that
provides this functionality in production, or
2. Use a different approach that relies only on methods available in the
standard Flink `SplitEnumeratorContext` API (e.g., tracking attempt numbers
separately within the enumerator state)
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -179,11 +181,53 @@ public void addSplitsBack(List<TieringSplit> splits, int
subtaskId) {
@Override
public void addReader(int subtaskId) {
LOG.info("Adding reader: {} to Tiering Source enumerator.", subtaskId);
- if (context.registeredReaders().containsKey(subtaskId)) {
+ Map<Integer, ReaderInfo> readerByAttempt =
+ context.registeredReadersOfAttempts().get(subtaskId);
+ if (readerByAttempt != null && !readerByAttempt.isEmpty()) {
readersAwaitingSplit.add(subtaskId);
+ int maxAttempt = max(readerByAttempt.keySet());
+ if (maxAttempt >= 1) {
+ if (isFailOvering) {
+ LOG.warn(
+ "Subtask {} (max attempt {}) registered during
ongoing failover.",
+ subtaskId,
+ maxAttempt);
+ } else {
+ LOG.warn(
+ "Detected failover: subtask {} has max attempt {}
> 0. Triggering global failover handling.",
+ subtaskId,
+ maxAttempt);
+ // should be failover
+ isFailOvering = true;
+ handleSourceReaderFailOver();
+ }
+
+ // if registered readers equal to current parallelism, check
whether all registered
+ // readers have same max maxAttempt
Review Comment:
Typo in comment: "same max maxAttempt" should be "same max attempt". The
word "max" is duplicated.
```suggestion
// readers have same max attempt
```
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -179,11 +181,53 @@ public void addSplitsBack(List<TieringSplit> splits, int
subtaskId) {
@Override
public void addReader(int subtaskId) {
LOG.info("Adding reader: {} to Tiering Source enumerator.", subtaskId);
- if (context.registeredReaders().containsKey(subtaskId)) {
+ Map<Integer, ReaderInfo> readerByAttempt =
+ context.registeredReadersOfAttempts().get(subtaskId);
+ if (readerByAttempt != null && !readerByAttempt.isEmpty()) {
readersAwaitingSplit.add(subtaskId);
+ int maxAttempt = max(readerByAttempt.keySet());
+ if (maxAttempt >= 1) {
+ if (isFailOvering) {
+ LOG.warn(
+ "Subtask {} (max attempt {}) registered during
ongoing failover.",
+ subtaskId,
+ maxAttempt);
+ } else {
+ LOG.warn(
+ "Detected failover: subtask {} has max attempt {}
> 0. Triggering global failover handling.",
+ subtaskId,
+ maxAttempt);
+ // should be failover
+ isFailOvering = true;
+ handleSourceReaderFailOver();
+ }
+
+ // if registered readers equal to current parallelism, check
whether all registered
+ // readers have same max maxAttempt
+ if (context.registeredReadersOfAttempts().size() ==
context.currentParallelism()) {
+ // Check if all readers have the same max attempt number
+ Set<Integer> maxAttempts =
+
context.registeredReadersOfAttempts().values().stream()
+ .map(_readerByAttempt ->
max(_readerByAttempt.keySet()))
+ .collect(Collectors.toSet());
+ int globalMaxAttempt = max(maxAttempts);
+ if (maxAttempts.size() == 1 && globalMaxAttempt >= 1) {
+ LOG.warn(
+ "Failover completed. All {} subtasks reached
the same attempt number {}. Current registered readers are {}",
+ context.currentParallelism(),
+ globalMaxAttempt,
+ context.registeredReadersOfAttempts());
Review Comment:
Same issue: calling `context.registeredReadersOfAttempts()` which doesn't
exist in the standard Flink `SplitEnumeratorContext` API.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -179,11 +181,53 @@ public void addSplitsBack(List<TieringSplit> splits, int
subtaskId) {
@Override
public void addReader(int subtaskId) {
LOG.info("Adding reader: {} to Tiering Source enumerator.", subtaskId);
- if (context.registeredReaders().containsKey(subtaskId)) {
+ Map<Integer, ReaderInfo> readerByAttempt =
+ context.registeredReadersOfAttempts().get(subtaskId);
+ if (readerByAttempt != null && !readerByAttempt.isEmpty()) {
readersAwaitingSplit.add(subtaskId);
+ int maxAttempt = max(readerByAttempt.keySet());
+ if (maxAttempt >= 1) {
+ if (isFailOvering) {
+ LOG.warn(
+ "Subtask {} (max attempt {}) registered during
ongoing failover.",
+ subtaskId,
+ maxAttempt);
+ } else {
+ LOG.warn(
+ "Detected failover: subtask {} has max attempt {}
> 0. Triggering global failover handling.",
+ subtaskId,
+ maxAttempt);
+ // should be failover
+ isFailOvering = true;
+ handleSourceReaderFailOver();
+ }
+
+ // if registered readers equal to current parallelism, check
whether all registered
+ // readers have same max maxAttempt
+ if (context.registeredReadersOfAttempts().size() ==
context.currentParallelism()) {
+ // Check if all readers have the same max attempt number
+ Set<Integer> maxAttempts =
+
context.registeredReadersOfAttempts().values().stream()
+ .map(_readerByAttempt ->
max(_readerByAttempt.keySet()))
+ .collect(Collectors.toSet());
+ int globalMaxAttempt = max(maxAttempts);
+ if (maxAttempts.size() == 1 && globalMaxAttempt >= 1) {
+ LOG.warn(
+ "Failover completed. All {} subtasks reached
the same attempt number {}. Current registered readers are {}",
+ context.currentParallelism(),
+ globalMaxAttempt,
+ context.registeredReadersOfAttempts());
Review Comment:
Same issue: calling `context.registeredReadersOfAttempts()` which doesn't
exist in the standard Flink `SplitEnumeratorContext` API.
```suggestion
ReaderInfo readerInfo = context.registeredReaders().get(subtaskId);
if (readerInfo != null) {
readersAwaitingSplit.add(subtaskId);
int attempt = readerInfo.getAttempt();
if (attempt >= 1) {
if (isFailOvering) {
LOG.warn(
"Subtask {} (attempt {}) registered during
ongoing failover.",
subtaskId,
attempt);
} else {
LOG.warn(
"Detected failover: subtask {} has attempt {} >
0. Triggering global failover handling.",
subtaskId,
attempt);
// should be failover
isFailOvering = true;
handleSourceReaderFailOver();
}
// if registered readers equal to current parallelism, check
whether all registered
// readers have same max attempt
if (context.registeredReaders().size() ==
context.currentParallelism()) {
// Check if all readers have the same attempt number
Set<Integer> attempts =
context.registeredReaders().values().stream()
.map(ReaderInfo::getAttempt)
.collect(Collectors.toSet());
int globalAttempt = max(attempts);
if (attempts.size() == 1 && globalAttempt >= 1) {
LOG.warn(
"Failover completed. All {} subtasks reached
the same attempt number {}. Current registered readers are {}",
context.currentParallelism(),
globalAttempt,
context.registeredReaders());
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java:
##########
@@ -0,0 +1,346 @@
+/*
+ * 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.flink.tiering;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.tiering.committer.CommittableMessageTypeInfo;
+import org.apache.fluss.flink.tiering.committer.TieringCommitOperatorFactory;
+import org.apache.fluss.flink.tiering.source.TableBucketWriteResultTypeInfo;
+import org.apache.fluss.flink.tiering.source.TieringSource;
+import org.apache.fluss.lake.values.ValuesLake;
+import org.apache.fluss.lake.values.tiering.ValuesLakeTieringFactory;
+import org.apache.fluss.lake.writer.LakeTieringFactory;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.TypeUtils;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.fluss.flink.tiering.source.TieringSource.TIERING_SOURCE_TRANSFORMATION_UID;
+import static
org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
+import static
org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test tiering failover. */
+class TieringFailoverITCase extends FlinkValuesTieringTestBase {
+ protected static final String DEFAULT_DB = "fluss";
+
+ private static StreamExecutionEnvironment execEnv;
+
+ private static final Schema pkSchema =
+ Schema.newBuilder()
+ .column("f_boolean", DataTypes.BOOLEAN())
+ .column("f_byte", DataTypes.TINYINT())
+ .column("f_short", DataTypes.SMALLINT())
+ .column("f_int", DataTypes.INT())
+ .column("f_long", DataTypes.BIGINT())
+ .column("f_float", DataTypes.FLOAT())
+ .column("f_double", DataTypes.DOUBLE())
+ .column("f_string", DataTypes.STRING())
+ .column("f_decimal1", DataTypes.DECIMAL(5, 2))
+ .column("f_decimal2", DataTypes.DECIMAL(20, 0))
+ .column("f_timestamp_ltz1", DataTypes.TIMESTAMP_LTZ(3))
+ .column("f_timestamp_ltz2", DataTypes.TIMESTAMP_LTZ(6))
+ .column("f_timestamp_ntz1", DataTypes.TIMESTAMP(3))
+ .column("f_timestamp_ntz2", DataTypes.TIMESTAMP(6))
+ .column("f_binary", DataTypes.BINARY(4))
+ .column("f_date", DataTypes.DATE())
+ .column("f_time", DataTypes.TIME())
+ .column("f_char", DataTypes.CHAR(3))
+ .column("f_bytes", DataTypes.BYTES())
+ .primaryKey("f_string")
+ .build();
+
+ @BeforeAll
+ protected static void beforeAll() {
+ FlinkValuesTieringTestBase.beforeAll();
+ execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ execEnv.setParallelism(2);
+ execEnv.enableCheckpointing(1000);
+ }
+
+ @Test
+ void testTiering() throws Exception {
+ // create a pk table, write some records and wait until snapshot
finished
+ TablePath t1 = TablePath.of(DEFAULT_DB, "pkTable");
+ long t1Id = createPkTable(t1, 1, false, pkSchema);
+ TableBucket t1Bucket = new TableBucket(t1Id, 0);
+ // write records
+ List<InternalRow> rows =
+ Arrays.asList(
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 1,
+ 1 + 400L,
+ 500.1f,
+ 600.0d,
+ "v1",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 2,
+ 2 + 400L,
+ 500.1f,
+ 600.0d,
+ "v2",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 3,
+ 3 + 400L,
+ 500.1f,
+ 600.0d,
+ "v3",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
+ List<InternalRow> expectedRows = new ArrayList<>(rows);
+ writeRows(t1, rows);
+ waitUntilSnapshot(t1Id, 1, 0);
+
+ // fail the first write to the pk table
+ ValuesLake.failWhen(t1.toString()).failWriteOnce();
+
+ // then start tiering job
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ // check the status of replica after synced
+ assertReplicaStatus(t1Bucket, 3);
+
+ checkDataInValuesPrimaryKeyTable(t1, rows);
+ // check snapshot property in values lake
+ Map<String, String> properties =
+ new HashMap<String, String>() {
+ {
+ put(
+ FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
+ "[{\"bucket\":0,\"offset\":3}]");
+ }
+ };
+ checkSnapshotPropertyInValues(t1, properties);
+
+ // then write data to the pk tables
+ // write records
+ rows =
+ Arrays.asList(
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 1,
+ 1 + 400L,
+ 500.1f,
+ 600.0d,
+ "v111",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 2,
+ 2 + 400L,
+ 500.1f,
+ 600.0d,
+ "v222",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 3,
+ 3 + 400L,
+ 500.1f,
+ 600.0d,
+ "v333",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}));
+ expectedRows.addAll(rows);
+ // write records
+ writeRows(t1, rows);
+
+ // check the status of replica of t1 after synced
+ // not check start offset since we won't
+ // update start log offset for primary key table
+ assertReplicaStatus(t1Bucket, expectedRows.size());
+
+ checkDataInValuesPrimaryKeyTable(t1, expectedRows);
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv)
throws Exception {
+ Configuration flussConfig = new Configuration(clientConf);
+ flussConfig.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500L));
+
+ LakeTieringFactory lakeTieringFactory = new ValuesLakeTieringFactory();
+
+ // build tiering source
+ TieringSource.Builder<?> tieringSourceBuilder =
+ new TieringSource.Builder<>(flussConfig, lakeTieringFactory);
+ if (flussConfig.get(POLL_TIERING_TABLE_INTERVAL) != null) {
+ tieringSourceBuilder.withPollTieringTableIntervalMs(
+ flussConfig.get(POLL_TIERING_TABLE_INTERVAL).toMillis());
+ }
+ TieringSource<?> tieringSource = tieringSourceBuilder.build();
+ DataStreamSource<?> source =
+ execEnv.fromSource(
+ tieringSource,
+ WatermarkStrategy.noWatermarks(),
+ "TieringSource",
+ TableBucketWriteResultTypeInfo.of(
+ () ->
lakeTieringFactory.getWriteResultSerializer()));
+
+ source.getTransformation().setUid(TIERING_SOURCE_TRANSFORMATION_UID);
+
+ source.transform(
+ "TieringCommitter",
+ CommittableMessageTypeInfo.of(
+ () ->
lakeTieringFactory.getCommittableSerializer()),
+ new TieringCommitOperatorFactory(flussConfig,
lakeTieringFactory))
+ .setParallelism(1)
+ .setMaxParallelism(1)
+ .sinkTo(new DiscardingSink())
+ .name("end")
+ .setParallelism(1);
+ String jobName =
+ execEnv.getConfiguration()
+ .getOptional(PipelineOptions.NAME)
+ .orElse("Fluss Lake Tiering FailOver IT Test.");
+
+ return execEnv.executeAsync(jobName);
+ }
+
+ private void checkDataInValuesPrimaryKeyTable(
+ TablePath tablePath, List<InternalRow> expectedRows) throws
Exception {
+ Iterator<InternalRow> acturalIterator =
getValuesRecords(tablePath).iterator();
Review Comment:
Typo in variable name: `acturalIterator` should be `actualIterator`.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/ValuesLake.java:
##########
@@ -0,0 +1,307 @@
+/*
+ * 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.lake.values;
+
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.utils.MapUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** {@link ValuesLake} for {@link ValuesLake}. */
+public class ValuesLake {
+ private static final Logger LOG =
LoggerFactory.getLogger(ValuesLake.class);
+
+ private static final Map<String, ValuesTable> globalTables =
MapUtils.newConcurrentHashMap();
+ private static final Map<String, TableFailureController>
FAILURE_CONTROLLERS =
+ MapUtils.newConcurrentHashMap();
+
+ public static TableFailureController failWhen(String tableId) {
+ return FAILURE_CONTROLLERS.computeIfAbsent(tableId, k -> new
TableFailureController());
+ }
+
+ public static void clearAllFailureControls() {
+ FAILURE_CONTROLLERS.clear();
+ }
+
+ public static Schema getTableSchema(String tableId) {
+ return globalTables.get(tableId).schema;
+ }
+
+ public static List<InternalRow> getResults(String tableId) {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
+ return table.getResult();
+ }
+
+ public static void writeRecord(String tableId, String stageId, LogRecord
record)
+ throws IOException {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
+ TableFailureController controller = FAILURE_CONTROLLERS.get(tableId);
+ if (controller != null) {
+ controller.checkWriteShouldFail(tableId);
+ }
+ table.writeRecord(stageId, record);
+ LOG.info("Write record to stage {}: {}", stageId, record);
+ }
+
+ public static long commit(
+ String tableId, List<String> stageIds, Map<String, String>
snapshotProperties)
+ throws IOException {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, "commit stage %s failed, table %s is not existed",
stageIds, tableId);
Review Comment:
Grammar error in error message: "is not existed" should be "does not exist".
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/ValuesLake.java:
##########
@@ -0,0 +1,307 @@
+/*
+ * 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.lake.values;
+
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.utils.MapUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** {@link ValuesLake} for {@link ValuesLake}. */
+public class ValuesLake {
+ private static final Logger LOG =
LoggerFactory.getLogger(ValuesLake.class);
+
+ private static final Map<String, ValuesTable> globalTables =
MapUtils.newConcurrentHashMap();
+ private static final Map<String, TableFailureController>
FAILURE_CONTROLLERS =
+ MapUtils.newConcurrentHashMap();
+
+ public static TableFailureController failWhen(String tableId) {
+ return FAILURE_CONTROLLERS.computeIfAbsent(tableId, k -> new
TableFailureController());
+ }
+
+ public static void clearAllFailureControls() {
+ FAILURE_CONTROLLERS.clear();
+ }
+
+ public static Schema getTableSchema(String tableId) {
+ return globalTables.get(tableId).schema;
+ }
+
+ public static List<InternalRow> getResults(String tableId) {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
+ return table.getResult();
+ }
+
+ public static void writeRecord(String tableId, String stageId, LogRecord
record)
+ throws IOException {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
Review Comment:
Grammar error in error message: "is not existed" should be "does not exist".
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -179,11 +181,53 @@ public void addSplitsBack(List<TieringSplit> splits, int
subtaskId) {
@Override
public void addReader(int subtaskId) {
LOG.info("Adding reader: {} to Tiering Source enumerator.", subtaskId);
- if (context.registeredReaders().containsKey(subtaskId)) {
+ Map<Integer, ReaderInfo> readerByAttempt =
+ context.registeredReadersOfAttempts().get(subtaskId);
+ if (readerByAttempt != null && !readerByAttempt.isEmpty()) {
readersAwaitingSplit.add(subtaskId);
+ int maxAttempt = max(readerByAttempt.keySet());
+ if (maxAttempt >= 1) {
+ if (isFailOvering) {
+ LOG.warn(
+ "Subtask {} (max attempt {}) registered during
ongoing failover.",
+ subtaskId,
+ maxAttempt);
+ } else {
+ LOG.warn(
+ "Detected failover: subtask {} has max attempt {}
> 0. Triggering global failover handling.",
+ subtaskId,
+ maxAttempt);
+ // should be failover
+ isFailOvering = true;
+ handleSourceReaderFailOver();
+ }
+
+ // if registered readers equal to current parallelism, check
whether all registered
+ // readers have same max maxAttempt
+ if (context.registeredReadersOfAttempts().size() ==
context.currentParallelism()) {
+ // Check if all readers have the same max attempt number
+ Set<Integer> maxAttempts =
+
context.registeredReadersOfAttempts().values().stream()
+ .map(_readerByAttempt ->
max(_readerByAttempt.keySet()))
+ .collect(Collectors.toSet());
+ int globalMaxAttempt = max(maxAttempts);
+ if (maxAttempts.size() == 1 && globalMaxAttempt >= 1) {
+ LOG.warn(
+ "Failover completed. All {} subtasks reached
the same attempt number {}. Current registered readers are {}",
+ context.currentParallelism(),
+ globalMaxAttempt,
+ context.registeredReadersOfAttempts());
Review Comment:
Same issue: calling `context.registeredReadersOfAttempts()` which doesn't
exist in the standard Flink `SplitEnumeratorContext` API.
```suggestion
ReaderInfo readerInfo = context.registeredReaders().get(subtaskId);
if (readerInfo != null) {
readersAwaitingSplit.add(subtaskId);
int attemptNumber = readerInfo.getAttemptNumber();
if (attemptNumber >= 1) {
if (isFailOvering) {
LOG.warn(
"Subtask {} (attempt {}) registered during
ongoing failover.",
subtaskId,
attemptNumber);
} else {
LOG.warn(
"Detected failover: subtask {} has attempt {} >
0. Triggering global failover handling.",
subtaskId,
attemptNumber);
// should be failover
isFailOvering = true;
handleSourceReaderFailOver();
}
// if registered readers equal to current parallelism, check
whether all registered
// readers have same max attempt
if (context.registeredReaders().size() ==
context.currentParallelism()) {
// Check if all readers have the same attempt number
Set<Integer> attemptNumbers =
context.registeredReaders().values().stream()
.map(ReaderInfo::getAttemptNumber)
.collect(Collectors.toSet());
int globalAttemptNumber = max(attemptNumbers);
if (attemptNumbers.size() == 1 && globalAttemptNumber >=
1) {
LOG.warn(
"Failover completed. All {} subtasks reached
the same attempt number {}. Current registered readers are {}",
context.currentParallelism(),
globalAttemptNumber,
context.registeredReaders());
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/FlussMockSplitEnumeratorContext.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * 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.flink.tiering.source.enumerator;
+
+import org.apache.fluss.utils.MapUtils;
+
+import org.apache.flink.api.connector.source.ReaderInfo;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A mock extends of {@link MockSplitEnumeratorContext} for testing purposes,
support registering
Review Comment:
Grammar error in documentation: "A mock extends of" should be "A mock
extension of".
```suggestion
* A mock extension of {@link MockSplitEnumeratorContext} for testing
purposes, support registering
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java:
##########
@@ -0,0 +1,346 @@
+/*
+ * 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.flink.tiering;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.tiering.committer.CommittableMessageTypeInfo;
+import org.apache.fluss.flink.tiering.committer.TieringCommitOperatorFactory;
+import org.apache.fluss.flink.tiering.source.TableBucketWriteResultTypeInfo;
+import org.apache.fluss.flink.tiering.source.TieringSource;
+import org.apache.fluss.lake.values.ValuesLake;
+import org.apache.fluss.lake.values.tiering.ValuesLakeTieringFactory;
+import org.apache.fluss.lake.writer.LakeTieringFactory;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.TypeUtils;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.fluss.flink.tiering.source.TieringSource.TIERING_SOURCE_TRANSFORMATION_UID;
+import static
org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
+import static
org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test tiering failover. */
+class TieringFailoverITCase extends FlinkValuesTieringTestBase {
+ protected static final String DEFAULT_DB = "fluss";
+
+ private static StreamExecutionEnvironment execEnv;
+
+ private static final Schema pkSchema =
+ Schema.newBuilder()
+ .column("f_boolean", DataTypes.BOOLEAN())
+ .column("f_byte", DataTypes.TINYINT())
+ .column("f_short", DataTypes.SMALLINT())
+ .column("f_int", DataTypes.INT())
+ .column("f_long", DataTypes.BIGINT())
+ .column("f_float", DataTypes.FLOAT())
+ .column("f_double", DataTypes.DOUBLE())
+ .column("f_string", DataTypes.STRING())
+ .column("f_decimal1", DataTypes.DECIMAL(5, 2))
+ .column("f_decimal2", DataTypes.DECIMAL(20, 0))
+ .column("f_timestamp_ltz1", DataTypes.TIMESTAMP_LTZ(3))
+ .column("f_timestamp_ltz2", DataTypes.TIMESTAMP_LTZ(6))
+ .column("f_timestamp_ntz1", DataTypes.TIMESTAMP(3))
+ .column("f_timestamp_ntz2", DataTypes.TIMESTAMP(6))
+ .column("f_binary", DataTypes.BINARY(4))
+ .column("f_date", DataTypes.DATE())
+ .column("f_time", DataTypes.TIME())
+ .column("f_char", DataTypes.CHAR(3))
+ .column("f_bytes", DataTypes.BYTES())
+ .primaryKey("f_string")
+ .build();
+
+ @BeforeAll
+ protected static void beforeAll() {
+ FlinkValuesTieringTestBase.beforeAll();
+ execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ execEnv.setParallelism(2);
+ execEnv.enableCheckpointing(1000);
+ }
+
+ @Test
+ void testTiering() throws Exception {
+ // create a pk table, write some records and wait until snapshot
finished
+ TablePath t1 = TablePath.of(DEFAULT_DB, "pkTable");
+ long t1Id = createPkTable(t1, 1, false, pkSchema);
+ TableBucket t1Bucket = new TableBucket(t1Id, 0);
+ // write records
+ List<InternalRow> rows =
+ Arrays.asList(
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 1,
+ 1 + 400L,
+ 500.1f,
+ 600.0d,
+ "v1",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 2,
+ 2 + 400L,
+ 500.1f,
+ 600.0d,
+ "v2",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 3,
+ 3 + 400L,
+ 500.1f,
+ 600.0d,
+ "v3",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
+ List<InternalRow> expectedRows = new ArrayList<>(rows);
+ writeRows(t1, rows);
+ waitUntilSnapshot(t1Id, 1, 0);
+
+ // fail the first write to the pk table
+ ValuesLake.failWhen(t1.toString()).failWriteOnce();
+
+ // then start tiering job
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ // check the status of replica after synced
+ assertReplicaStatus(t1Bucket, 3);
+
+ checkDataInValuesPrimaryKeyTable(t1, rows);
+ // check snapshot property in values lake
+ Map<String, String> properties =
+ new HashMap<String, String>() {
+ {
+ put(
+ FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
+ "[{\"bucket\":0,\"offset\":3}]");
+ }
+ };
+ checkSnapshotPropertyInValues(t1, properties);
+
+ // then write data to the pk tables
+ // write records
+ rows =
+ Arrays.asList(
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 1,
+ 1 + 400L,
+ 500.1f,
+ 600.0d,
+ "v111",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 2,
+ 2 + 400L,
+ 500.1f,
+ 600.0d,
+ "v222",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 3,
+ 3 + 400L,
+ 500.1f,
+ 600.0d,
+ "v333",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}));
+ expectedRows.addAll(rows);
+ // write records
+ writeRows(t1, rows);
+
+ // check the status of replica of t1 after synced
+ // not check start offset since we won't
+ // update start log offset for primary key table
+ assertReplicaStatus(t1Bucket, expectedRows.size());
+
+ checkDataInValuesPrimaryKeyTable(t1, expectedRows);
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv)
throws Exception {
+ Configuration flussConfig = new Configuration(clientConf);
+ flussConfig.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500L));
+
+ LakeTieringFactory lakeTieringFactory = new ValuesLakeTieringFactory();
+
+ // build tiering source
+ TieringSource.Builder<?> tieringSourceBuilder =
+ new TieringSource.Builder<>(flussConfig, lakeTieringFactory);
+ if (flussConfig.get(POLL_TIERING_TABLE_INTERVAL) != null) {
+ tieringSourceBuilder.withPollTieringTableIntervalMs(
+ flussConfig.get(POLL_TIERING_TABLE_INTERVAL).toMillis());
+ }
+ TieringSource<?> tieringSource = tieringSourceBuilder.build();
+ DataStreamSource<?> source =
+ execEnv.fromSource(
+ tieringSource,
+ WatermarkStrategy.noWatermarks(),
+ "TieringSource",
+ TableBucketWriteResultTypeInfo.of(
+ () ->
lakeTieringFactory.getWriteResultSerializer()));
+
+ source.getTransformation().setUid(TIERING_SOURCE_TRANSFORMATION_UID);
+
+ source.transform(
+ "TieringCommitter",
+ CommittableMessageTypeInfo.of(
+ () ->
lakeTieringFactory.getCommittableSerializer()),
+ new TieringCommitOperatorFactory(flussConfig,
lakeTieringFactory))
+ .setParallelism(1)
+ .setMaxParallelism(1)
+ .sinkTo(new DiscardingSink())
+ .name("end")
+ .setParallelism(1);
+ String jobName =
+ execEnv.getConfiguration()
+ .getOptional(PipelineOptions.NAME)
+ .orElse("Fluss Lake Tiering FailOver IT Test.");
+
+ return execEnv.executeAsync(jobName);
+ }
+
+ private void checkDataInValuesPrimaryKeyTable(
+ TablePath tablePath, List<InternalRow> expectedRows) throws
Exception {
+ Iterator<InternalRow> acturalIterator =
getValuesRecords(tablePath).iterator();
+ Iterator<InternalRow> iterator = expectedRows.iterator();
+ while (iterator.hasNext() && acturalIterator.hasNext()) {
Review Comment:
Same typo: `acturalIterator` should be `actualIterator`.
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/ValuesLake.java:
##########
@@ -0,0 +1,307 @@
+/*
+ * 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.lake.values;
+
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.utils.MapUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** {@link ValuesLake} for {@link ValuesLake}. */
Review Comment:
Incorrect Javadoc: The class-level documentation says "{@link ValuesLake}
for {@link ValuesLake}" which is redundant and doesn't provide meaningful
information. It should describe the purpose of the class, e.g., "In-memory
implementation of a lake storage for testing purposes."
```suggestion
/**
* In-memory implementation of a lake storage for testing purposes.
* <p>
* Provides utilities for managing tables, writing records, committing
stages,
* and retrieving results in a test environment.
*/
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/ValuesLake.java:
##########
@@ -0,0 +1,307 @@
+/*
+ * 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.lake.values;
+
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.utils.MapUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** {@link ValuesLake} for {@link ValuesLake}. */
+public class ValuesLake {
+ private static final Logger LOG =
LoggerFactory.getLogger(ValuesLake.class);
+
+ private static final Map<String, ValuesTable> globalTables =
MapUtils.newConcurrentHashMap();
+ private static final Map<String, TableFailureController>
FAILURE_CONTROLLERS =
+ MapUtils.newConcurrentHashMap();
+
+ public static TableFailureController failWhen(String tableId) {
+ return FAILURE_CONTROLLERS.computeIfAbsent(tableId, k -> new
TableFailureController());
+ }
+
+ public static void clearAllFailureControls() {
+ FAILURE_CONTROLLERS.clear();
+ }
+
+ public static Schema getTableSchema(String tableId) {
+ return globalTables.get(tableId).schema;
+ }
+
+ public static List<InternalRow> getResults(String tableId) {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
+ return table.getResult();
+ }
+
+ public static void writeRecord(String tableId, String stageId, LogRecord
record)
+ throws IOException {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
Review Comment:
Grammar error in error message: "is not existed" should be "does not exist".
```suggestion
checkNotNull(table, tableId + " does not exist");
return table.getResult();
}
public static void writeRecord(String tableId, String stageId, LogRecord
record)
throws IOException {
ValuesTable table = globalTables.get(tableId);
checkNotNull(table, tableId + " does not exist");
```
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/lake/values/ValuesLake.java:
##########
@@ -0,0 +1,307 @@
+/*
+ * 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.lake.values;
+
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.utils.MapUtils;
+import org.apache.fluss.utils.types.Tuple2;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** {@link ValuesLake} for {@link ValuesLake}. */
+public class ValuesLake {
+ private static final Logger LOG =
LoggerFactory.getLogger(ValuesLake.class);
+
+ private static final Map<String, ValuesTable> globalTables =
MapUtils.newConcurrentHashMap();
+ private static final Map<String, TableFailureController>
FAILURE_CONTROLLERS =
+ MapUtils.newConcurrentHashMap();
+
+ public static TableFailureController failWhen(String tableId) {
+ return FAILURE_CONTROLLERS.computeIfAbsent(tableId, k -> new
TableFailureController());
+ }
+
+ public static void clearAllFailureControls() {
+ FAILURE_CONTROLLERS.clear();
+ }
+
+ public static Schema getTableSchema(String tableId) {
+ return globalTables.get(tableId).schema;
+ }
+
+ public static List<InternalRow> getResults(String tableId) {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
+ return table.getResult();
+ }
+
+ public static void writeRecord(String tableId, String stageId, LogRecord
record)
+ throws IOException {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, tableId + " is not existed");
+ TableFailureController controller = FAILURE_CONTROLLERS.get(tableId);
+ if (controller != null) {
+ controller.checkWriteShouldFail(tableId);
+ }
+ table.writeRecord(stageId, record);
+ LOG.info("Write record to stage {}: {}", stageId, record);
+ }
+
+ public static long commit(
+ String tableId, List<String> stageIds, Map<String, String>
snapshotProperties)
+ throws IOException {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(table, "commit stage %s failed, table %s is not existed",
stageIds, tableId);
+ table.commit(stageIds, snapshotProperties);
+ LOG.info("Commit table {} stage {}", tableId, stageIds);
+ return table.getSnapshotId();
+ }
+
+ public static void abort(String tableId, List<String> stageIds) {
+ ValuesTable table = globalTables.get(tableId);
+ checkNotNull(
+ table, "abort stage record %s failed, table %s is not
existed", stageIds, tableId);
Review Comment:
Grammar error in error message: "is not existed" should be "does not exist".
##########
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/TieringFailoverITCase.java:
##########
@@ -0,0 +1,346 @@
+/*
+ * 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.flink.tiering;
+
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.flink.tiering.committer.CommittableMessageTypeInfo;
+import org.apache.fluss.flink.tiering.committer.TieringCommitOperatorFactory;
+import org.apache.fluss.flink.tiering.source.TableBucketWriteResultTypeInfo;
+import org.apache.fluss.flink.tiering.source.TieringSource;
+import org.apache.fluss.lake.values.ValuesLake;
+import org.apache.fluss.lake.values.tiering.ValuesLakeTieringFactory;
+import org.apache.fluss.lake.writer.LakeTieringFactory;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.Decimal;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.utils.TypeUtils;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import static
org.apache.fluss.flink.tiering.source.TieringSource.TIERING_SOURCE_TRANSFORMATION_UID;
+import static
org.apache.fluss.flink.tiering.source.TieringSourceOptions.POLL_TIERING_TABLE_INTERVAL;
+import static
org.apache.fluss.lake.committer.BucketOffset.FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test tiering failover. */
+class TieringFailoverITCase extends FlinkValuesTieringTestBase {
+ protected static final String DEFAULT_DB = "fluss";
+
+ private static StreamExecutionEnvironment execEnv;
+
+ private static final Schema pkSchema =
+ Schema.newBuilder()
+ .column("f_boolean", DataTypes.BOOLEAN())
+ .column("f_byte", DataTypes.TINYINT())
+ .column("f_short", DataTypes.SMALLINT())
+ .column("f_int", DataTypes.INT())
+ .column("f_long", DataTypes.BIGINT())
+ .column("f_float", DataTypes.FLOAT())
+ .column("f_double", DataTypes.DOUBLE())
+ .column("f_string", DataTypes.STRING())
+ .column("f_decimal1", DataTypes.DECIMAL(5, 2))
+ .column("f_decimal2", DataTypes.DECIMAL(20, 0))
+ .column("f_timestamp_ltz1", DataTypes.TIMESTAMP_LTZ(3))
+ .column("f_timestamp_ltz2", DataTypes.TIMESTAMP_LTZ(6))
+ .column("f_timestamp_ntz1", DataTypes.TIMESTAMP(3))
+ .column("f_timestamp_ntz2", DataTypes.TIMESTAMP(6))
+ .column("f_binary", DataTypes.BINARY(4))
+ .column("f_date", DataTypes.DATE())
+ .column("f_time", DataTypes.TIME())
+ .column("f_char", DataTypes.CHAR(3))
+ .column("f_bytes", DataTypes.BYTES())
+ .primaryKey("f_string")
+ .build();
+
+ @BeforeAll
+ protected static void beforeAll() {
+ FlinkValuesTieringTestBase.beforeAll();
+ execEnv = StreamExecutionEnvironment.getExecutionEnvironment();
+ execEnv.setParallelism(2);
+ execEnv.enableCheckpointing(1000);
+ }
+
+ @Test
+ void testTiering() throws Exception {
+ // create a pk table, write some records and wait until snapshot
finished
+ TablePath t1 = TablePath.of(DEFAULT_DB, "pkTable");
+ long t1Id = createPkTable(t1, 1, false, pkSchema);
+ TableBucket t1Bucket = new TableBucket(t1Id, 0);
+ // write records
+ List<InternalRow> rows =
+ Arrays.asList(
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 1,
+ 1 + 400L,
+ 500.1f,
+ 600.0d,
+ "v1",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 2,
+ 2 + 400L,
+ 500.1f,
+ 600.0d,
+ "v2",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 3,
+ 3 + 400L,
+ 500.1f,
+ 600.0d,
+ "v3",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+ TimestampLtz.fromEpochMillis(1698235273400L),
+ TimestampLtz.fromEpochMillis(1698235273400L,
7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L, 8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
+ List<InternalRow> expectedRows = new ArrayList<>(rows);
+ writeRows(t1, rows);
+ waitUntilSnapshot(t1Id, 1, 0);
+
+ // fail the first write to the pk table
+ ValuesLake.failWhen(t1.toString()).failWriteOnce();
+
+ // then start tiering job
+ JobClient jobClient = buildTieringJob(execEnv);
+ try {
+ // check the status of replica after synced
+ assertReplicaStatus(t1Bucket, 3);
+
+ checkDataInValuesPrimaryKeyTable(t1, rows);
+ // check snapshot property in values lake
+ Map<String, String> properties =
+ new HashMap<String, String>() {
+ {
+ put(
+ FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY,
+ "[{\"bucket\":0,\"offset\":3}]");
+ }
+ };
+ checkSnapshotPropertyInValues(t1, properties);
+
+ // then write data to the pk tables
+ // write records
+ rows =
+ Arrays.asList(
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 1,
+ 1 + 400L,
+ 500.1f,
+ 600.0d,
+ "v111",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 2,
+ 2 + 400L,
+ 500.1f,
+ 600.0d,
+ "v222",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}),
+ row(
+ true,
+ (byte) 100,
+ (short) 200,
+ 3,
+ 3 + 400L,
+ 500.1f,
+ 600.0d,
+ "v333",
+ Decimal.fromUnscaledLong(900, 5, 2),
+ Decimal.fromBigDecimal(new
java.math.BigDecimal(1000), 20, 0),
+
TimestampLtz.fromEpochMillis(1698235273400L),
+
TimestampLtz.fromEpochMillis(1698235273400L, 7000),
+ TimestampNtz.fromMillis(1698235273501L),
+ TimestampNtz.fromMillis(1698235273501L,
8000),
+ new byte[] {5, 6, 7, 8},
+ TypeUtils.castFromString("2023-10-25",
DataTypes.DATE()),
+ TypeUtils.castFromString("09:30:00.0",
DataTypes.TIME()),
+ BinaryString.fromString("abc"),
+ new byte[] {1, 2, 3, 4, 5, 6, 7, 8, 9,
10}));
+ expectedRows.addAll(rows);
+ // write records
+ writeRows(t1, rows);
+
+ // check the status of replica of t1 after synced
+ // not check start offset since we won't
+ // update start log offset for primary key table
+ assertReplicaStatus(t1Bucket, expectedRows.size());
+
+ checkDataInValuesPrimaryKeyTable(t1, expectedRows);
+ } finally {
+ jobClient.cancel().get();
+ }
+ }
+
+ protected JobClient buildTieringJob(StreamExecutionEnvironment execEnv)
throws Exception {
+ Configuration flussConfig = new Configuration(clientConf);
+ flussConfig.set(POLL_TIERING_TABLE_INTERVAL, Duration.ofMillis(500L));
+
+ LakeTieringFactory lakeTieringFactory = new ValuesLakeTieringFactory();
+
+ // build tiering source
+ TieringSource.Builder<?> tieringSourceBuilder =
+ new TieringSource.Builder<>(flussConfig, lakeTieringFactory);
+ if (flussConfig.get(POLL_TIERING_TABLE_INTERVAL) != null) {
+ tieringSourceBuilder.withPollTieringTableIntervalMs(
+ flussConfig.get(POLL_TIERING_TABLE_INTERVAL).toMillis());
+ }
+ TieringSource<?> tieringSource = tieringSourceBuilder.build();
+ DataStreamSource<?> source =
+ execEnv.fromSource(
+ tieringSource,
+ WatermarkStrategy.noWatermarks(),
+ "TieringSource",
+ TableBucketWriteResultTypeInfo.of(
+ () ->
lakeTieringFactory.getWriteResultSerializer()));
+
+ source.getTransformation().setUid(TIERING_SOURCE_TRANSFORMATION_UID);
+
+ source.transform(
+ "TieringCommitter",
+ CommittableMessageTypeInfo.of(
+ () ->
lakeTieringFactory.getCommittableSerializer()),
+ new TieringCommitOperatorFactory(flussConfig,
lakeTieringFactory))
+ .setParallelism(1)
+ .setMaxParallelism(1)
+ .sinkTo(new DiscardingSink())
+ .name("end")
+ .setParallelism(1);
+ String jobName =
+ execEnv.getConfiguration()
+ .getOptional(PipelineOptions.NAME)
+ .orElse("Fluss Lake Tiering FailOver IT Test.");
+
+ return execEnv.executeAsync(jobName);
+ }
+
+ private void checkDataInValuesPrimaryKeyTable(
+ TablePath tablePath, List<InternalRow> expectedRows) throws
Exception {
+ Iterator<InternalRow> acturalIterator =
getValuesRecords(tablePath).iterator();
+ Iterator<InternalRow> iterator = expectedRows.iterator();
+ while (iterator.hasNext() && acturalIterator.hasNext()) {
+ InternalRow row = iterator.next();
+ InternalRow record = acturalIterator.next();
+ assertThat(record.getBoolean(0)).isEqualTo(row.getBoolean(0));
+ assertThat(record.getByte(1)).isEqualTo(row.getByte(1));
+ assertThat(record.getShort(2)).isEqualTo(row.getShort(2));
+ assertThat(record.getInt(3)).isEqualTo(row.getInt(3));
+ assertThat(record.getLong(4)).isEqualTo(row.getLong(4));
+ assertThat(record.getFloat(5)).isEqualTo(row.getFloat(5));
+ assertThat(record.getDouble(6)).isEqualTo(row.getDouble(6));
+ assertThat(record.getString(7)).isEqualTo(row.getString(7));
+ assertThat(record.getDecimal(8, 5, 2)).isEqualTo(row.getDecimal(8,
5, 2));
+ assertThat(record.getDecimal(9, 20,
0)).isEqualTo(row.getDecimal(9, 20, 0));
+ assertThat(record.getTimestampLtz(10,
3)).isEqualTo(row.getTimestampLtz(10, 3));
+ assertThat(record.getTimestampLtz(11,
6)).isEqualTo(row.getTimestampLtz(11, 6));
+ assertThat(record.getTimestampNtz(12,
6)).isEqualTo(row.getTimestampNtz(12, 6));
+ assertThat(record.getTimestampNtz(13,
6)).isEqualTo(row.getTimestampNtz(13, 6));
+ assertThat(record.getBinary(14, 4)).isEqualTo(row.getBinary(14,
4));
+ assertThat(record.getInt(15)).isEqualTo(row.getInt(15));
+ assertThat(record.getInt(16)).isEqualTo(row.getInt(16));
+ assertThat(record.getChar(17, 3)).isEqualTo(row.getChar(17, 3));
+ assertThat(record.getBinary(18, 10)).isEqualTo(row.getBinary(18,
10));
+ }
+ assertThat(acturalIterator.hasNext()).isFalse();
Review Comment:
Same typo: `acturalIterator` should be `actualIterator`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
