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/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 3ec19cd5 [FLINK-27875] Introduce ResetRowKindRecordIterator to reset
INSERT row kind
3ec19cd5 is described below
commit 3ec19cd53e7db287028c142fb5754d73353ff1af
Author: JingsongLi <[email protected]>
AuthorDate: Fri Jun 3 08:37:36 2022 +0800
[FLINK-27875] Introduce ResetRowKindRecordIterator to reset INSERT row kind
---
...erator.java => ResetRowKindRecordIterator.java} | 39 +++++++++-------------
.../source/ValueContentRowDataRecordIterator.java | 27 ++++-----------
.../source/ValueCountRowDataRecordIterator.java | 24 ++++---------
.../source/RowDataRecordIteratorTestBase.java | 29 +++++++++++++++-
4 files changed, 56 insertions(+), 63 deletions(-)
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ResetRowKindRecordIterator.java
similarity index 56%
copy from
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
copy to
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ResetRowKindRecordIterator.java
index 3f8b7d81..e32ffaf6 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ResetRowKindRecordIterator.java
@@ -25,40 +25,33 @@ import org.apache.flink.types.RowKind;
import java.io.IOException;
-/** A {@link RecordReader.RecordIterator} mapping a {@link KeyValue} to its
value. */
-public class ValueContentRowDataRecordIterator implements
RecordReader.RecordIterator<RowData> {
+/**
+ * A {@link RecordReader.RecordIterator} which resets {@link RowKind#INSERT}
to previous key value.
+ */
+public abstract class ResetRowKindRecordIterator implements
RecordReader.RecordIterator<RowData> {
private final RecordReader.RecordIterator<KeyValue> kvIterator;
- public
ValueContentRowDataRecordIterator(RecordReader.RecordIterator<KeyValue>
kvIterator) {
+ private KeyValue keyValue;
+
+ public ResetRowKindRecordIterator(RecordReader.RecordIterator<KeyValue>
kvIterator) {
this.kvIterator = kvIterator;
}
- @Override
- public RowData next() throws IOException {
- KeyValue kv = kvIterator.next();
- if (kv == null) {
- return null;
+ public final KeyValue nextKeyValue() throws IOException {
+ // The RowData is reused in kvIterator, we should set back to insert
kind
+ // Failure to do so will result in uncontrollable exceptions
+ if (keyValue != null) {
+ keyValue.key().setRowKind(RowKind.INSERT);
+ keyValue.value().setRowKind(RowKind.INSERT);
}
- RowData rowData = kv.value();
- // kv.value() is reused, so we need to set row kind each time
- switch (kv.valueKind()) {
- case ADD:
- rowData.setRowKind(RowKind.INSERT);
- break;
- case DELETE:
- rowData.setRowKind(RowKind.DELETE);
- break;
- default:
- throw new UnsupportedOperationException(
- "Unknown value kind " + kv.valueKind().name());
- }
- return rowData;
+ keyValue = kvIterator.next();
+ return keyValue;
}
@Override
- public void releaseBatch() {
+ public final void releaseBatch() {
kvIterator.releaseBatch();
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
index 3f8b7d81..716f4838 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueContentRowDataRecordIterator.java
@@ -20,45 +20,30 @@ package org.apache.flink.table.store.table.source;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.store.file.KeyValue;
+import org.apache.flink.table.store.file.ValueKind;
import org.apache.flink.table.store.file.utils.RecordReader;
import org.apache.flink.types.RowKind;
import java.io.IOException;
/** A {@link RecordReader.RecordIterator} mapping a {@link KeyValue} to its
value. */
-public class ValueContentRowDataRecordIterator implements
RecordReader.RecordIterator<RowData> {
-
- private final RecordReader.RecordIterator<KeyValue> kvIterator;
+public class ValueContentRowDataRecordIterator extends
ResetRowKindRecordIterator {
public
ValueContentRowDataRecordIterator(RecordReader.RecordIterator<KeyValue>
kvIterator) {
- this.kvIterator = kvIterator;
+ super(kvIterator);
}
@Override
public RowData next() throws IOException {
- KeyValue kv = kvIterator.next();
+ KeyValue kv = nextKeyValue();
if (kv == null) {
return null;
}
RowData rowData = kv.value();
- // kv.value() is reused, so we need to set row kind each time
- switch (kv.valueKind()) {
- case ADD:
- rowData.setRowKind(RowKind.INSERT);
- break;
- case DELETE:
- rowData.setRowKind(RowKind.DELETE);
- break;
- default:
- throw new UnsupportedOperationException(
- "Unknown value kind " + kv.valueKind().name());
+ if (kv.valueKind() == ValueKind.DELETE) {
+ rowData.setRowKind(RowKind.DELETE);
}
return rowData;
}
-
- @Override
- public void releaseBatch() {
- kvIterator.releaseBatch();
- }
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueCountRowDataRecordIterator.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueCountRowDataRecordIterator.java
index 1bb2f2a5..1d37cc0d 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueCountRowDataRecordIterator.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/ValueCountRowDataRecordIterator.java
@@ -33,9 +33,8 @@ import java.io.IOException;
* according to its key. These {@link RowData}s are the same. The number of
rows depends on the
* value of {@link KeyValue}.
*/
-public class ValueCountRowDataRecordIterator implements
RecordReader.RecordIterator<RowData> {
+public class ValueCountRowDataRecordIterator extends
ResetRowKindRecordIterator {
- private final RecordReader.RecordIterator<KeyValue> kvIterator;
private final @Nullable ProjectedRowData projectedRowData;
private RowData rowData;
@@ -43,7 +42,7 @@ public class ValueCountRowDataRecordIterator implements
RecordReader.RecordItera
public ValueCountRowDataRecordIterator(
RecordReader.RecordIterator<KeyValue> kvIterator, @Nullable
int[][] projection) {
- this.kvIterator = kvIterator;
+ super(kvIterator);
this.projectedRowData = projection == null ? null :
ProjectedRowData.from(projection);
this.rowData = null;
@@ -57,30 +56,19 @@ public class ValueCountRowDataRecordIterator implements
RecordReader.RecordItera
count--;
return rowData;
} else {
- KeyValue kv = kvIterator.next();
+ KeyValue kv = nextKeyValue();
if (kv == null) {
return null;
}
- if (projectedRowData == null) {
- rowData = kv.key();
- } else {
- rowData = projectedRowData.replaceRow(kv.key());
- }
+ rowData =
+ projectedRowData == null ? kv.key() :
projectedRowData.replaceRow(kv.key());
long value = kv.value().getLong(0);
- // kv.value() is reused, so we need to set row kind each time
- if (value > 0) {
- rowData.setRowKind(RowKind.INSERT);
- } else {
+ if (value < 0) {
rowData.setRowKind(RowKind.DELETE);
}
count = Math.abs(value);
}
}
}
-
- @Override
- public void releaseBatch() {
- kvIterator.releaseBatch();
- }
}
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/RowDataRecordIteratorTestBase.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/RowDataRecordIteratorTestBase.java
index e422514d..f33df366 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/RowDataRecordIteratorTestBase.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/table/source/RowDataRecordIteratorTestBase.java
@@ -23,11 +23,15 @@ import org.apache.flink.table.store.file.KeyValue;
import org.apache.flink.table.store.file.utils.RecordReader;
import org.apache.flink.table.store.file.utils.ReusingTestData;
import org.apache.flink.table.store.file.utils.TestReusingRecordReader;
+import org.apache.flink.types.RowKind;
+import java.io.IOException;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Function;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Tests for {@link RecordReader.RecordIterator} of {@link RowData}. */
public abstract class RowDataRecordIteratorTestBase {
@@ -44,8 +48,31 @@ public abstract class RowDataRecordIteratorTestBase {
if (kvIterator == null) {
break;
}
+
+ RecordReader.RecordIterator<KeyValue> assertKvIterator =
+ new RecordReader.RecordIterator<KeyValue>() {
+
+ KeyValue previous;
+
+ @Override
+ public KeyValue next() throws IOException {
+ // check
+ if (previous != null) {
+
assertThat(previous.key().getRowKind()).isEqualTo(RowKind.INSERT);
+
assertThat(previous.value().getRowKind()).isEqualTo(RowKind.INSERT);
+ }
+ previous = kvIterator.next();
+ return previous;
+ }
+
+ @Override
+ public void releaseBatch() {
+ kvIterator.releaseBatch();
+ }
+ };
+
RecordReader.RecordIterator<RowData> rowDataIterator =
- rowDataIteratorSupplier.apply(kvIterator);
+ rowDataIteratorSupplier.apply(assertKvIterator);
RowData rowData;
while (true) {
rowData = rowDataIterator.next();
