You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/01/10 06:18:29 UTC
(incubator-paimon) branch master updated: [cdc] Minor improve CdcUnawareBucketSink and fix unstable test (#2666)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f3b37355b [cdc] Minor improve CdcUnawareBucketSink and fix unstable test (#2666)
f3b37355b is described below
commit f3b37355bd2dcf075dfa2ccdaed1fba26e6fc814
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Wed Jan 10 14:18:24 2024 +0800
[cdc] Minor improve CdcUnawareBucketSink and fix unstable test (#2666)
---
.../paimon/flink/sink/cdc/CdcSinkBuilder.java | 2 +-
...ketWriteSink.java => CdcUnawareBucketSink.java} | 19 ++-----
.../sink/cdc/CdcUnawareBucketWriteOperator.java | 62 ++--------------------
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 3 +-
.../sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java | 51 +++++-------------
.../sink/cdc/FlinkCdcSyncTableSinkITCase.java | 53 ++++++------------
.../apache/paimon/flink/sink/cdc/TestTable.java | 50 +++++++++++------
7 files changed, 74 insertions(+), 166 deletions(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
index 56dfc2b63..380cdffb8 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
@@ -135,7 +135,7 @@ public class CdcSinkBuilder<T> {
private DataStreamSink<?> buildForUnawareBucket(DataStream<CdcRecord> parsed) {
FileStoreTable dataTable = (FileStoreTable) table;
- return new CdcUnawareBucketWriteSink((AppendOnlyFileStoreTable) dataTable, parallelism)
+ return new CdcUnawareBucketSink((AppendOnlyFileStoreTable) dataTable, parallelism)
.sinkFrom(parsed);
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteSink.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
similarity index 68%
rename from paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteSink.java
rename to paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
index d4d4b4fd9..5c6c1d363 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteSink.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketSink.java
@@ -18,33 +18,22 @@
package org.apache.paimon.flink.sink.cdc;
-import org.apache.paimon.flink.sink.LogSinkFunction;
+import org.apache.paimon.flink.sink.Committable;
import org.apache.paimon.flink.sink.StoreSinkWrite;
import org.apache.paimon.flink.sink.UnawareBucketSink;
import org.apache.paimon.table.AppendOnlyFileStoreTable;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import java.util.Map;
-
/** CDC Sink for unaware bucket table. */
-public class CdcUnawareBucketWriteSink extends UnawareBucketSink {
+public class CdcUnawareBucketSink extends UnawareBucketSink<CdcRecord> {
- public CdcUnawareBucketWriteSink(AppendOnlyFileStoreTable table, Integer parallelism) {
+ public CdcUnawareBucketSink(AppendOnlyFileStoreTable table, Integer parallelism) {
super(table, null, null, parallelism, false);
}
- public CdcUnawareBucketWriteSink(
- AppendOnlyFileStoreTable table,
- Integer parallelism,
- Map<String, String> overwritePartitions,
- LogSinkFunction logSinkFunction,
- boolean boundedInput) {
- super(table, overwritePartitions, logSinkFunction, parallelism, boundedInput);
- }
-
@Override
- protected OneInputStreamOperator createWriteOperator(
+ protected OneInputStreamOperator<CdcRecord, Committable> createWriteOperator(
StoreSinkWrite.Provider writeProvider, String commitUser) {
return new CdcUnawareBucketWriteOperator(table, writeProvider, commitUser);
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java
index fdaf93775..c57a40f3f 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/CdcUnawareBucketWriteOperator.java
@@ -18,80 +18,28 @@
package org.apache.paimon.flink.sink.cdc;
-import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.sink.PrepareCommitOperator;
import org.apache.paimon.flink.sink.StoreSinkWrite;
-import org.apache.paimon.flink.sink.TableWriteOperator;
-import org.apache.paimon.options.ConfigOption;
-import org.apache.paimon.options.ConfigOptions;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowKind;
-import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import java.io.IOException;
-import java.time.Duration;
-import java.util.Optional;
-
-import static org.apache.paimon.flink.sink.cdc.CdcRecordUtils.toGenericRow;
-
-/** A {@link PrepareCommitOperator} to write {@link CdcRecord} with bucket. */
-public class CdcUnawareBucketWriteOperator extends TableWriteOperator<CdcRecord> {
-
- private static final long serialVersionUID = 1L;
-
- public static final ConfigOption<Duration> RETRY_SLEEP_TIME =
- ConfigOptions.key("cdc.retry-sleep-time")
- .durationType()
- .defaultValue(Duration.ofMillis(500));
-
- protected final long retrySleepMillis;
+/** A {@link PrepareCommitOperator} to write {@link CdcRecord} to unaware-bucket mode table. */
+public class CdcUnawareBucketWriteOperator extends CdcRecordStoreWriteOperator {
public CdcUnawareBucketWriteOperator(
FileStoreTable table,
StoreSinkWrite.Provider storeSinkWriteProvider,
String initialCommitUser) {
super(table, storeSinkWriteProvider, initialCommitUser);
- this.retrySleepMillis =
- table.coreOptions().toConfiguration().get(RETRY_SLEEP_TIME).toMillis();
- }
-
- @Override
- public void initializeState(StateInitializationContext context) throws Exception {
- table = table.copyWithLatestSchema();
- super.initializeState(context);
- }
-
- @Override
- protected boolean containLogSystem() {
- return false;
}
@Override
public void processElement(StreamRecord<CdcRecord> element) throws Exception {
- CdcRecord record = element.getValue();
- Optional<GenericRow> optionalConverted = toGenericRow(record, table.schema().fields());
- if (!optionalConverted.isPresent()) {
- while (true) {
- table = table.copyWithLatestSchema();
- optionalConverted = toGenericRow(record, table.schema().fields());
- if (optionalConverted.isPresent()) {
- break;
- }
- Thread.sleep(retrySleepMillis);
- }
- write.replace(table);
- }
-
- // Only insert type messages are processed
- try {
- GenericRow cdcDataRow = optionalConverted.get();
- if (cdcDataRow.getRowKind() == RowKind.INSERT) {
- write.write(cdcDataRow);
- }
- } catch (Exception e) {
- throw new IOException(e);
+ // only accepts INSERT record
+ if (element.getValue().kind() == RowKind.INSERT) {
+ super.processElement(element);
}
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index bb3933b32..c20aaf612 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -171,8 +171,7 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
}
private void buildForUnawareBucket(FileStoreTable table, DataStream<CdcRecord> parsed) {
- new CdcUnawareBucketWriteSink((AppendOnlyFileStoreTable) table, parallelism)
- .sinkFrom(parsed);
+ new CdcUnawareBucketSink((AppendOnlyFileStoreTable) table, parallelism).sinkFrom(parsed);
}
private void buildDividedCdcSink() {
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
index b7aca65ca..6080dd9df 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkITCase.java
@@ -21,7 +21,6 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogUtils;
-import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.util.AbstractTestBase;
import org.apache.paimon.fs.FileIO;
@@ -29,15 +28,12 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
-import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
import org.apache.paimon.utils.TraceableFileIO;
@@ -55,7 +51,6 @@ import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_PARALLELISM;
@@ -86,7 +81,7 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
innerTestRandomCdcEvents(() -> -1, true);
}
- private void innerTestRandomCdcEvents(Supplier<Integer> bucket, boolean isAppendTable)
+ private void innerTestRandomCdcEvents(Supplier<Integer> bucket, boolean unawareBucketMode)
throws Exception {
ThreadLocalRandom random = ThreadLocalRandom.current();
@@ -111,7 +106,7 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
random.nextInt(maxSchemaChanges) + 1,
random.nextInt(maxPartitions) + 1,
random.nextInt(maxKeys) + 1,
- isAppendTable);
+ unawareBucketMode);
testTables.add(testTable);
Path tablePath;
@@ -143,9 +138,8 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
fileIO,
testTable.initialRowType(),
Collections.singletonList("pt"),
- Arrays.asList("pt", "k"),
- bucket.get(),
- isAppendTable);
+ unawareBucketMode ? Collections.emptyList() : Arrays.asList("pt", "k"),
+ bucket.get());
fileStoreTables.add(fileStoreTable);
}
@@ -179,32 +173,17 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
// enable failure when running jobs if needed
FailingFileIO.reset(failingName, 10, 10000);
-
- if (!isAppendTable) {
- env.execute();
- } else {
- // When the unaware bucket table is synchronized, it creates a topology and compacts, so
- // the task does not terminate.In the test task, we are given enough time to execute the
- // job.
+ if (unawareBucketMode) {
+ // there's a compact operator which won't terminate
env.executeAsync();
- long waitResultTime = TimeUnit.SECONDS.toMillis(60);
- Thread.sleep(waitResultTime);
+ } else {
+ env.execute();
}
// no failure when checking results
FailingFileIO.reset(failingName, 0, 1);
-
- for (int i = 0; i < numTables; i++) {
- FileStoreTable table = fileStoreTables.get(i).copyWithLatestSchema();
- SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
- TableSchema schema = schemaManager.latest().get();
-
- ReadBuilder readBuilder = table.newReadBuilder();
- TableScan.Plan plan = readBuilder.newScan().plan();
- try (RecordReaderIterator<InternalRow> it =
- new RecordReaderIterator<>(readBuilder.newRead().createReader(plan))) {
- testTables.get(i).assertResult(schema, it);
- }
+ for (int i = 0; i < fileStoreTables.size(); i++) {
+ testTables.get(i).assertResult(fileStoreTables.get(i));
}
}
@@ -214,8 +193,7 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
RowType rowType,
List<String> partitions,
List<String> primaryKeys,
- int numBucket,
- boolean isAppendTable)
+ int numBucket)
throws Exception {
Options conf = new Options();
conf.set(CoreOptions.BUCKET, numBucket);
@@ -226,12 +204,7 @@ public class FlinkCdcSyncDatabaseSinkITCase extends AbstractTestBase {
TableSchema tableSchema =
SchemaUtils.forceCommit(
new SchemaManager(fileIO, tablePath),
- new Schema(
- rowType.getFields(),
- partitions,
- isAppendTable ? Collections.emptyList() : primaryKeys,
- conf.toMap(),
- ""));
+ new Schema(rowType.getFields(), partitions, primaryKeys, conf.toMap(), ""));
return FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
index 2be7098ec..a47561741 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
@@ -22,7 +22,6 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.data.InternalRow;
import org.apache.paimon.flink.FlinkCatalogFactory;
import org.apache.paimon.flink.util.AbstractTestBase;
import org.apache.paimon.fs.FileIO;
@@ -30,15 +29,12 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
-import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
import org.apache.paimon.utils.TraceableFileIO;
@@ -56,7 +52,6 @@ import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.concurrent.TimeUnit;
/** IT cases for {@link CdcSinkBuilder}. */
public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
@@ -91,8 +86,8 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
innerTestRandomCdcEvents(-1, false, true);
}
- private void innerTestRandomCdcEvents(int numBucket, boolean globalIndex, boolean isAppendTable)
- throws Exception {
+ private void innerTestRandomCdcEvents(
+ int numBucket, boolean globalIndex, boolean unawareBucketMode) throws Exception {
ThreadLocalRandom random = ThreadLocalRandom.current();
int numEvents = random.nextInt(1500) + 1;
@@ -108,7 +103,7 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
numSchemaChanges,
numPartitions,
numKeys,
- isAppendTable);
+ unawareBucketMode);
Path tablePath;
FileIO fileIO;
@@ -134,15 +129,18 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
// no failure when creating table
FailingFileIO.reset(failingName, 0, 1);
+ List<String> primaryKeys =
+ unawareBucketMode
+ ? Collections.emptyList()
+ : globalIndex ? Collections.singletonList("k") : Arrays.asList("pt", "k");
FileStoreTable table =
createFileStoreTable(
tablePath,
fileIO,
testTable.initialRowType(),
Collections.singletonList("pt"),
- globalIndex ? Collections.singletonList("k") : Arrays.asList("pt", "k"),
- numBucket,
- isAppendTable);
+ primaryKeys,
+ numBucket);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getCheckpointConfig().setCheckpointInterval(100);
@@ -171,30 +169,17 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
// enable failure when running jobs if needed
FailingFileIO.reset(failingName, 10, 10000);
- if (!isAppendTable) {
- env.execute();
- } else {
- // When the unaware bucket table is synchronized, it creates a topology and compacts, so
- // the task does not terminate.In the test task, we are given enough time to execute the
- // job.
+ if (unawareBucketMode) {
+ // there's a compact operator which won't terminate
env.executeAsync();
- long waitResultTime = TimeUnit.SECONDS.toMillis(30);
- Thread.sleep(waitResultTime);
+ } else {
+ env.execute();
}
// no failure when checking results
FailingFileIO.reset(failingName, 0, 1);
- table = table.copyWithLatestSchema();
- SchemaManager schemaManager = new SchemaManager(table.fileIO(), table.location());
- TableSchema schema = schemaManager.latest().get();
-
- ReadBuilder readBuilder = table.newReadBuilder();
- TableScan.Plan plan = readBuilder.newScan().plan();
- try (RecordReaderIterator<InternalRow> it =
- new RecordReaderIterator<>(readBuilder.newRead().createReader(plan))) {
- testTable.assertResult(schema, it);
- }
+ testTable.assertResult(table);
}
private FileStoreTable createFileStoreTable(
@@ -203,8 +188,7 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
RowType rowType,
List<String> partitions,
List<String> primaryKeys,
- int numBucket,
- boolean isAppendTable)
+ int numBucket)
throws Exception {
Options conf = new Options();
conf.set(CoreOptions.BUCKET, numBucket);
@@ -215,12 +199,7 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
TableSchema tableSchema =
SchemaUtils.forceCommit(
new SchemaManager(fileIO, tablePath),
- new Schema(
- rowType.getFields(),
- partitions,
- isAppendTable ? Collections.emptyList() : primaryKeys,
- conf.toMap(),
- ""));
+ new Schema(rowType.getFields(), partitions, primaryKeys, conf.toMap(), ""));
return FileStoreTableFactory.create(fileIO, tablePath, tableSchema);
}
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java
index 525a05096..beb8786d1 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/TestTable.java
@@ -19,7 +19,11 @@
package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.reader.RecordReaderIterator;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.TableScan;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
@@ -29,7 +33,6 @@ import org.apache.paimon.types.RowType;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -185,23 +188,40 @@ public class TestTable {
return events;
}
- public void assertResult(TableSchema schema, Iterator<InternalRow> it) {
+ public void assertResult(FileStoreTable table) throws Exception {
Map<Integer, Map<String, String>> actual = new HashMap<>();
- while (it.hasNext()) {
- InternalRow row = it.next();
- Map<String, String> fields = new HashMap<>();
- for (int i = 0; i < schema.fieldNames().size(); i++) {
- if (!row.isNullAt(i)) {
- fields.put(
- schema.fieldNames().get(i),
- String.valueOf(
- schema.fields().get(i).type().equals(DataTypes.BIGINT())
- ? row.getLong(i)
- : row.getInt(i)));
+ while (true) {
+ actual.clear();
+ table = table.copyWithLatestSchema();
+ TableSchema schema = table.schema();
+ ReadBuilder readBuilder = table.newReadBuilder();
+ TableScan.Plan plan = readBuilder.newScan().plan();
+
+ try (RecordReaderIterator<InternalRow> it =
+ new RecordReaderIterator<>(readBuilder.newRead().createReader(plan))) {
+ while (it.hasNext()) {
+ InternalRow row = it.next();
+ Map<String, String> fields = new HashMap<>();
+ for (int i = 0; i < schema.fieldNames().size(); i++) {
+ if (!row.isNullAt(i)) {
+ fields.put(
+ schema.fieldNames().get(i),
+ String.valueOf(
+ schema.fields().get(i).type().equals(DataTypes.BIGINT())
+ ? row.getLong(i)
+ : row.getInt(i)));
+ }
+ }
+ actual.put(Integer.valueOf(fields.get("k")), fields);
}
}
- actual.put(Integer.valueOf(fields.get("k")), fields);
+
+ if (actual.size() == expected.size()) {
+ assertThat(actual).isEqualTo(expected);
+ break;
+ }
+
+ Thread.sleep(500);
}
- assertThat(actual).isEqualTo(expected);
}
}