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 b4f187f3 [FLINK-29070] Provide a option to force the removal of the
normalize node when streaming read
b4f187f3 is described below
commit b4f187f370d3e47c436bc9c0aafa5f6e381e6387
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Aug 23 11:11:49 2022 +0800
[FLINK-29070] Provide a option to force the removal of the normalize node
when streaming read
This closes #273
---
docs/layouts/shortcodes/generated/core_configuration.html | 6 ++++++
.../flink/table/store/connector/source/TableStoreSource.java | 5 +++++
.../src/main/java/org/apache/flink/table/store/CoreOptions.java | 9 +++++++++
3 files changed, 20 insertions(+)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 36eea4d7..b093a1a4 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -110,6 +110,12 @@
<td><p>Enum</p></td>
<td>Specify the startup mode for log consumer.<br /><br />Possible
values:<ul><li>"full": Perform a snapshot on the table upon first startup, and
continue to read the latest changes.</li><li>"latest": Start from the
latest.</li><li>"from-timestamp": Start from user-supplied
timestamp.</li></ul></td>
</tr>
+ <tr>
+ <td><h5>log.scan.remove-normalize</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to force the removal of the normalize node when
streaming read. Note: This is dangerous and is likely to cause data errors if
downstream is used to calculate aggregation and the input is not complete
changelog.</td>
+ </tr>
<tr>
<td><h5>log.scan.timestamp-millis</h5></td>
<td style="word-wrap: break-word;">(none)</td>
diff --git
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
index 2969f980..3404308a 100644
---
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
+++
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/TableStoreSource.java
@@ -55,6 +55,7 @@ import java.util.List;
import static org.apache.flink.table.store.CoreOptions.CHANGELOG_PRODUCER;
import static org.apache.flink.table.store.CoreOptions.LOG_CHANGELOG_MODE;
import static org.apache.flink.table.store.CoreOptions.LOG_CONSISTENCY;
+import static
org.apache.flink.table.store.CoreOptions.LOG_SCAN_REMOVE_NORMALIZE;
/**
* Table source to create {@link FileStoreSource} under batch mode or
change-tracking is disabled.
@@ -134,6 +135,10 @@ public class TableStoreSource
} else if (table instanceof ChangelogWithKeyFileStoreTable) {
Configuration options =
Configuration.fromMap(table.schema().options());
+ if (options.get(LOG_SCAN_REMOVE_NORMALIZE)) {
+ return ChangelogMode.all();
+ }
+
if (logStoreTableFactory == null
&& options.get(CHANGELOG_PRODUCER) !=
ChangelogProducer.NONE) {
return ChangelogMode.all();
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
index 21769cdf..10626a70 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/CoreOptions.java
@@ -303,6 +303,15 @@ public class CoreOptions implements Serializable {
.defaultValue(LogChangelogMode.AUTO)
.withDescription("Specify the log changelog mode for
table.");
+ public static final ConfigOption<Boolean> LOG_SCAN_REMOVE_NORMALIZE =
+ ConfigOptions.key("log.scan.remove-normalize")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Whether to force the removal of the normalize
node when streaming read."
+ + " Note: This is dangerous and is likely
to cause data errors if downstream"
+ + " is used to calculate aggregation and
the input is not complete changelog.");
+
public static final ConfigOption<String> LOG_KEY_FORMAT =
ConfigOptions.key("log.key.format")
.stringType()
