You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "Alibaba-HZY (via GitHub)" <gi...@apache.org> on 2023/04/02 16:04:42 UTC
[GitHub] [incubator-paimon] Alibaba-HZY opened a new pull request, #812: [core]Add streaming read from option (#778)
Alibaba-HZY opened a new pull request, #812:
URL: https://github.com/apache/incubator-paimon/pull/812
Purpose
The user can determine the reading mode of the external log table by configuring the streaming-read-from option. The default value is file-store, which will read data from the file system. If 'streaming-read-from' = 'log-store', data will be read from the message queue. #778
Tests
LogSystemITCase
API and Format
CoreOptions:
New: streaming-read-from
Documentation
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161658878
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamingReadMode> STREAMING_READ_MODE =
+ key("streaming-read-mode")
+ .enumType(StreamingReadMode.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
Review Comment:
```suggestion
.text("The mode of streaming read that specifies to read the data of table file or log")
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161660269
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1054,6 +1079,49 @@ public static FileFormatType fromValue(String value) {
}
}
+ /** Specifies the type for streaming read. */
+ public enum StreamingReadMode implements DescribedEnum {
+ LOG("log", "Read from log system such as kafka."),
+ FILE("file", "Read from file store.");
Review Comment:
```suggestion
FILE("file", "Reads from the file store.");
```
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1054,6 +1079,49 @@ public static FileFormatType fromValue(String value) {
}
}
+ /** Specifies the type for streaming read. */
+ public enum StreamingReadMode implements DescribedEnum {
+ LOG("log", "Read from log system such as kafka."),
Review Comment:
```suggestion
LOG("log", "Reads from the log store."),
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161659729
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamingReadMode> STREAMING_READ_MODE =
+ key("streaming-read-mode")
+ .enumType(StreamingReadMode.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
+ .linebreak()
+ .linebreak()
+ .text("Possible values:")
+ .linebreak()
+ .list(
+ text(
+ StreamingReadMode.FILE.getValue()
+ + ": Will read from the file store"))
+ .list(
+ text(
+ StreamingReadMode.LOG.getValue()
+ + ":Will read from the log store, the user must configure log.system"))
Review Comment:
```suggestion
+ ": Read from the data of table log store."))
```
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamingReadMode> STREAMING_READ_MODE =
+ key("streaming-read-mode")
+ .enumType(StreamingReadMode.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
+ .linebreak()
+ .linebreak()
+ .text("Possible values:")
+ .linebreak()
+ .list(
+ text(
+ StreamingReadMode.FILE.getValue()
+ + ": Will read from the file store"))
Review Comment:
```suggestion
+ ": Reads from the data of table file store."))
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161667218
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java:
##########
@@ -132,6 +141,10 @@ private static void validateFileStoreContinuous(Options options) {
throw new ValidationException(
"File store continuous reading dose not support eventual consistency mode.");
}
+ if (streamingReadMode == StreamingReadMode.LOG) {
+ throw new ValidationException(
Review Comment:
Add the test case for this.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on pull request #812: [core]Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1498603563
> See paimon-docs/README.md.
I have executed mvn package -Pgenerate-docs -pl paimon-docs -nsu -DskipTests
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159507030
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,23 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamReadType> LOG_READ_FROM =
+ key("log.read-from")
+ .enumType(StreamReadType.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
+ .linebreak()
+ .linebreak()
+ .text("Possible values:")
+ .linebreak()
+ .list(text("\"file\": Will read from the file store"))
+ .list(
+ text(
+ "\"log-system\":Will read from the file store, the user must configure log.system"))
Review Comment:
```suggestion
"\"log-system\":Will read from the log store"))
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1156898337
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -559,6 +559,23 @@ public class CoreOptions implements Serializable {
.withDescription(
"Full compaction will be constantly triggered after delta commits.");
+ public static final ConfigOption<StreamReadType> STREAMING_READ_FROM =
+ key("streaming-read-from")
+ .enumType(StreamReadType.class)
+ .defaultValue(StreamReadType.FILE_STORE)
Review Comment:
default AUTO?
log system configured: choose log-system.
not configured: choose file file.
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1045,6 +1066,45 @@ public static FileFormatType fromValue(String value) {
}
}
+ /** Specifies the type for streaming read. */
+ public enum StreamReadType implements DescribedEnum {
+ LOG_STORE("log-store", "Read from log store such as kafka."),
Review Comment:
`log-system` and `file`?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161493559
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamReadType> STREAMING_READ_FROM =
+ key("streaming-read-from")
Review Comment:
Could we use `streaming-read-mode` and the value of this option could be `FILE` and `LOG`?
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1054,6 +1079,49 @@ public static FileFormatType fromValue(String value) {
}
}
+ /** Specifies the type for streaming read. */
+ public enum StreamReadType implements DescribedEnum {
Review Comment:
```suggestion
public enum StreamingReadMode implements DescribedEnum {
```
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -142,7 +144,10 @@ private ContinuousFileStoreSource buildContinuousFileSource() {
// TODO visit all options through CoreOptions
StartupMode startupMode = CoreOptions.startupMode(conf);
- if (logSourceProvider == null) {
+ StreamReadType streamReadType = CoreOptions.streamReadType(conf);
+
+ // no log system or
+ if (logSourceProvider == null || streamReadType == FILE) {
Review Comment:
@Alibaba-HZY, it's better to set the streaming read mode as log via `log.system` and this could be updated to `streamingReadMode == LOG`.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on pull request #812: [core] Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1505000889
> Thanks you,I commit too many times before and I squash, so I do git push -f.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161666935
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java:
##########
@@ -58,7 +58,7 @@ public class FlinkConnectorOptions {
.list(
TextElement.text(
"\"kafka\": Kafka log system, the data is double written to file"
- + " store and kafka, and the streaming read will be read from kafka."))
+ + " store and kafka, the user can configure streaming-read-mode to determine the store of streaming reads. For example, streaming-read-mode=log will read data from kafka and streaming-read-mode=file will read from the file store."))
Review Comment:
```suggestion
+ " store and kafka, and the streaming read will be read from kafka. If streaming read from file, configures streaming-read-mode to file."))
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161659348
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamingReadMode> STREAMING_READ_MODE =
+ key("streaming-read-mode")
+ .enumType(StreamingReadMode.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
+ .linebreak()
+ .linebreak()
+ .text("Possible values:")
+ .linebreak()
+ .list(
+ text(
+ StreamingReadMode.FILE.getValue()
+ + ": Will read from the file store"))
Review Comment:
```suggestion
+ ": Reads from the data of table file."))
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on pull request #812: [core]Add streaming read from option (#778)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1498507185
See paimon-docs/README.md.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159740807
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -142,7 +144,10 @@ private ContinuousFileStoreSource buildContinuousFileSource() {
// TODO visit all options through CoreOptions
StartupMode startupMode = CoreOptions.startupMode(conf);
- if (logSourceProvider == null) {
+ StreamReadType streamReadType = CoreOptions.streamReadType(conf);
+
+ // no log system or
+ if (logSourceProvider == null || streamReadType == FILE) {
Review Comment:
no log system or log system is configured but streamReadType is FILE
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159542867
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LogSystemITCase.java:
##########
@@ -67,4 +67,99 @@ public void testAppendOnlyWithEventual() throws Exception {
write.getJobClient().get().cancel();
read.close();
}
+
+ @Test
+ public void testReadFromFile() throws Exception {
+ createTopicIfNotExists("test-double-sink", 1);
+ // disable checkpointing to test eventual
+ env.getCheckpointConfig().disableCheckpointing();
+ env.setParallelism(1);
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE kafka_file_double_sink (\n"
+ + " word STRING ,\n"
+ + " cnt BIGINT,\n"
+ + " PRIMARY KEY (word) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'changelog-producer' = 'full-compaction',\n"
+ + " 'log.consistency' = 'eventual',\n"
+ + " 'log.system' = 'kafka',\n"
+ + " 'log.read-from'='file',\n"
+ + " 'fields.cnt.aggregate-function' = 'sum',\n"
+ + " 'kafka.bootstrap.servers' = '%s',\n"
+ + " 'kafka.topic' = 'test-double-sink',\n"
+ + " 'kafka.transaction.timeout.ms'='30000'\n"
+ + "\n"
+ + ");",
+ getBootstrapServers()));
+ tEnv.executeSql(
+ "CREATE TEMPORARY TABLE word_table (\n"
+ + " word STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'fields.word.length' = '1'\n"
+ + ");");
+ TableResult write =
+ tEnv.executeSql(
+ "INSERT INTO kafka_file_double_sink SELECT word, COUNT(*) FROM word_table GROUP BY word;");
+ BlockingIterator<Row, Row> read =
+ BlockingIterator.of(
+ tEnv.executeSql("SELECT * FROM kafka_file_double_sink").collect());
+ List<Row> collect = read.collect(10);
+ assertThat(collect).hasSize(10);
Review Comment:
@Alibaba-HZY, could this verify the actual data content? IMO the size of actual data isn't enough.
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LogSystemITCase.java:
##########
@@ -67,4 +67,99 @@ public void testAppendOnlyWithEventual() throws Exception {
write.getJobClient().get().cancel();
read.close();
}
+
+ @Test
+ public void testReadFromFile() throws Exception {
+ createTopicIfNotExists("test-double-sink", 1);
+ // disable checkpointing to test eventual
+ env.getCheckpointConfig().disableCheckpointing();
+ env.setParallelism(1);
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE kafka_file_double_sink (\n"
+ + " word STRING ,\n"
+ + " cnt BIGINT,\n"
+ + " PRIMARY KEY (word) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'changelog-producer' = 'full-compaction',\n"
+ + " 'log.consistency' = 'eventual',\n"
+ + " 'log.system' = 'kafka',\n"
+ + " 'log.read-from'='file',\n"
+ + " 'fields.cnt.aggregate-function' = 'sum',\n"
+ + " 'kafka.bootstrap.servers' = '%s',\n"
+ + " 'kafka.topic' = 'test-double-sink',\n"
+ + " 'kafka.transaction.timeout.ms'='30000'\n"
+ + "\n"
+ + ");",
+ getBootstrapServers()));
+ tEnv.executeSql(
+ "CREATE TEMPORARY TABLE word_table (\n"
+ + " word STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'fields.word.length' = '1'\n"
+ + ");");
+ TableResult write =
+ tEnv.executeSql(
+ "INSERT INTO kafka_file_double_sink SELECT word, COUNT(*) FROM word_table GROUP BY word;");
+ BlockingIterator<Row, Row> read =
+ BlockingIterator.of(
+ tEnv.executeSql("SELECT * FROM kafka_file_double_sink").collect());
+ List<Row> collect = read.collect(10);
+ assertThat(collect).hasSize(10);
Review Comment:
@Alibaba-HZY, could this verify the actual data content? The size of actual data isn't enough.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161658878
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamingReadMode> STREAMING_READ_MODE =
+ key("streaming-read-mode")
+ .enumType(StreamingReadMode.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
Review Comment:
```suggestion
.text("The mode of streaming read that specifies to read the data of table File or Log")
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159713013
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -142,7 +144,10 @@ private ContinuousFileStoreSource buildContinuousFileSource() {
// TODO visit all options through CoreOptions
StartupMode startupMode = CoreOptions.startupMode(conf);
- if (logSourceProvider == null) {
+ StreamReadType streamReadType = CoreOptions.streamReadType(conf);
+
+ // no log system or
+ if (logSourceProvider == null || streamReadType == FILE) {
Review Comment:
no log system or log system is configured but streamReadType is FILE
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1156907431
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -559,6 +559,23 @@ public class CoreOptions implements Serializable {
.withDescription(
"Full compaction will be constantly triggered after delta commits.");
+ public static final ConfigOption<StreamReadType> STREAMING_READ_FROM =
+ key("streaming-read-from")
+ .enumType(StreamReadType.class)
+ .defaultValue(StreamReadType.FILE_STORE)
Review Comment:
ok
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -559,6 +559,23 @@ public class CoreOptions implements Serializable {
.withDescription(
"Full compaction will be constantly triggered after delta commits.");
+ public static final ConfigOption<StreamReadType> STREAMING_READ_FROM =
+ key("streaming-read-from")
Review Comment:
ok
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161665606
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java:
##########
@@ -132,6 +141,10 @@ private static void validateFileStoreContinuous(Options options) {
throw new ValidationException(
"File store continuous reading dose not support eventual consistency mode.");
Review Comment:
```suggestion
"File store continuous reading does not support eventual consistency mode.");
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161653118
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,27 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamReadType> STREAMING_READ_FROM =
+ key("streaming-read-from")
Review Comment:
have been modified
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159542867
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/LogSystemITCase.java:
##########
@@ -67,4 +67,99 @@ public void testAppendOnlyWithEventual() throws Exception {
write.getJobClient().get().cancel();
read.close();
}
+
+ @Test
+ public void testReadFromFile() throws Exception {
+ createTopicIfNotExists("test-double-sink", 1);
+ // disable checkpointing to test eventual
+ env.getCheckpointConfig().disableCheckpointing();
+ env.setParallelism(1);
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE kafka_file_double_sink (\n"
+ + " word STRING ,\n"
+ + " cnt BIGINT,\n"
+ + " PRIMARY KEY (word) NOT ENFORCED\n"
+ + ")\n"
+ + "WITH (\n"
+ + " 'merge-engine' = 'aggregation',\n"
+ + " 'changelog-producer' = 'full-compaction',\n"
+ + " 'log.consistency' = 'eventual',\n"
+ + " 'log.system' = 'kafka',\n"
+ + " 'log.read-from'='file',\n"
+ + " 'fields.cnt.aggregate-function' = 'sum',\n"
+ + " 'kafka.bootstrap.servers' = '%s',\n"
+ + " 'kafka.topic' = 'test-double-sink',\n"
+ + " 'kafka.transaction.timeout.ms'='30000'\n"
+ + "\n"
+ + ");",
+ getBootstrapServers()));
+ tEnv.executeSql(
+ "CREATE TEMPORARY TABLE word_table (\n"
+ + " word STRING\n"
+ + ") WITH (\n"
+ + " 'connector' = 'datagen',\n"
+ + " 'fields.word.length' = '1'\n"
+ + ");");
+ TableResult write =
+ tEnv.executeSql(
+ "INSERT INTO kafka_file_double_sink SELECT word, COUNT(*) FROM word_table GROUP BY word;");
+ BlockingIterator<Row, Row> read =
+ BlockingIterator.of(
+ tEnv.executeSql("SELECT * FROM kafka_file_double_sink").collect());
+ List<Row> collect = read.collect(10);
+ assertThat(collect).hasSize(10);
Review Comment:
Could this verify the actual data content?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on pull request #812: [core] Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1504943648
> Hi @Alibaba-HZY , why did many force pushing? Is this ready to review now?
Previously we set streaming-read-mode default value according to whether log.system is set, and then determine whether to read mq or file only according to this parameter, but when the user executes sql with hint, the parameter option is UnmodifiableMap, it cannot be added to the value. So I changed the if logic. If the user does not configure streamingReadMode but the parameter contains log.system then to read mq, but does not contain it then read file, if the streamingReadMode is configured, the decision is made based on it.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on pull request #812: [core] Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1504946295
> Hi @Alibaba-HZY , why did many force pushing? Is this ready to review now?
I have opened the action in my clone project, and this push' workflows has been approved
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on pull request #812: [core]Add streaming read from option (#778)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1504469516
@Alibaba-HZY Can you open action for your cloned project?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159541902
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -142,7 +144,10 @@ private ContinuousFileStoreSource buildContinuousFileSource() {
// TODO visit all options through CoreOptions
StartupMode startupMode = CoreOptions.startupMode(conf);
- if (logSourceProvider == null) {
+ StreamReadType streamReadType = CoreOptions.streamReadType(conf);
+
+ // no log system or
+ if (logSourceProvider == null || streamReadType == FILE) {
Review Comment:
When `logSourceProvider` isn't null and `streamReadType` is `FILE`, is it right to build continuous file source like `latest` startup mode?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159523341
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,23 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamReadType> LOG_READ_FROM =
+ key("log.read-from")
Review Comment:
It's better to name `streaming-read-from` because `log.read-from` is something confusing for users, in which the fact is that log reads from log system.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1156899441
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -559,6 +559,23 @@ public class CoreOptions implements Serializable {
.withDescription(
"Full compaction will be constantly triggered after delta commits.");
+ public static final ConfigOption<StreamReadType> STREAMING_READ_FROM =
+ key("streaming-read-from")
Review Comment:
`log.read-from`?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161665689
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java:
##########
@@ -113,16 +115,23 @@ static Optional<LogStoreTableFactory> createOptionalLogStoreFactory(
options.forEach(configOptions::setString);
if (configOptions.get(LOG_SYSTEM).equalsIgnoreCase(NONE)) {
- // Use file store continuous reading
+ // Use file store continuous reading and set streaming-read-mode
+ options.put(STREAMING_READ_MODE.key(), StreamingReadMode.FILE.getValue());
validateFileStoreContinuous(configOptions);
return Optional.empty();
+ } else {
+ // user does not set streaming-read-mode, making default value log
+ if (!options.containsKey(STREAMING_READ_MODE.key())) {
+ options.put(STREAMING_READ_MODE.key(), StreamingReadMode.LOG.getValue());
+ }
}
return Optional.of(discoverLogStoreFactory(classLoader, configOptions.get(LOG_SYSTEM)));
}
private static void validateFileStoreContinuous(Options options) {
LogChangelogMode changelogMode = options.get(LOG_CHANGELOG_MODE);
+ StreamingReadMode streamingReadMode = options.get(STREAMING_READ_MODE);
if (changelogMode == LogChangelogMode.UPSERT) {
throw new ValidationException(
"File store continuous reading dose not support upsert changelog mode.");
Review Comment:
```suggestion
"File store continuous reading does not support upsert changelog mode.");
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161664269
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java:
##########
@@ -132,6 +141,10 @@ private static void validateFileStoreContinuous(Options options) {
throw new ValidationException(
"File store continuous reading dose not support eventual consistency mode.");
}
+ if (streamingReadMode == StreamingReadMode.LOG) {
+ throw new ValidationException(
+ "Table data is stored only in the file store, can not reading from the log store.");
Review Comment:
```suggestion
"File store continuous reading does not support the log streaming read mode.");
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas closed pull request #812: [core] Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas closed pull request #812: [core] Add streaming read from option (#778)
URL: https://github.com/apache/incubator-paimon/pull/812
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] sonarcloud[bot] commented on pull request #812: [core]Add streaming read from option (#778)
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1501550056
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_incubator-paimon&pullRequest=812)
[](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=CODE_SMELL) [0 Code Smells](https://sonarcloud.io/project/issues?id=apache_incubator-paimon&pullRequest=812&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_incubator-paimon&pullRequest=812) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_incubator-paimon&pullRequest=812&metric=new_duplicated_lines_density&view=list) [0.0% Duplication](https://sonarcloud.io/component_measures?id=apache_incubator-paimon&pullRequest=812&metric=new_duplicated_lines_density&view=list)
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on pull request #812: [core] Add streaming read from option (#778)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1504996412
> > Hi @Alibaba-HZY , why did many force pushing? Is this ready to review now?
>
> Previously we set streaming-read-mode default value according to whether log.system is set, and then determine whether to read mq or file only according to this parameter, but when the user executes sql with hint, the parameter option is UnmodifiableMap, it cannot be added to the value. So I changed the if logic. If the user does not configure streamingReadMode but the parameter contains log.system then to read mq, but does not contain it then read file, if the streamingReadMode is configured, the decision is made based on it.
Thanks for the information, I suggest you directly push a commit to avoid force pushing.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas merged pull request #812: [core] Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas merged PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on pull request #812: [core] Add streaming read from option (#778)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#issuecomment-1504927738
Hi @Alibaba-HZY , why did many force pushing? Is this ready to review now?
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161664269
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java:
##########
@@ -132,6 +141,10 @@ private static void validateFileStoreContinuous(Options options) {
throw new ValidationException(
"File store continuous reading dose not support eventual consistency mode.");
}
+ if (streamingReadMode == StreamingReadMode.LOG) {
+ throw new ValidationException(
+ "Table data is stored only in the file store, can not reading from the log store.");
Review Comment:
```suggestion
"File store continuous reading dose not support the log streaming read mode.");
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1161660109
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -1054,6 +1079,49 @@ public static FileFormatType fromValue(String value) {
}
}
+ /** Specifies the type for streaming read. */
+ public enum StreamingReadMode implements DescribedEnum {
+ LOG("log", "Read from log system such as kafka."),
Review Comment:
```suggestion
LOG("log", "Reads from log store like kafka etc."),
```
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] SteNicholas commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "SteNicholas (via GitHub)" <gi...@apache.org>.
SteNicholas commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159524354
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,23 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamReadType> LOG_READ_FROM =
+ key("log.read-from")
+ .enumType(StreamReadType.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
+ .linebreak()
+ .linebreak()
+ .text("Possible values:")
+ .linebreak()
+ .list(text("\"file\": Will read from the file store"))
Review Comment:
Use the `StreamReadType.FILE.getValue()`.
##########
paimon-core/src/main/java/org/apache/paimon/CoreOptions.java:
##########
@@ -568,6 +568,23 @@ public class CoreOptions implements Serializable {
"Only used to force TableScan to construct 'ContinuousCompactorStartingScanner' and "
+ "'ContinuousCompactorFollowUpScanner' for dedicated streaming compaction job.");
+ public static final ConfigOption<StreamReadType> LOG_READ_FROM =
+ key("log.read-from")
+ .enumType(StreamReadType.class)
+ .noDefaultValue()
+ .withDescription(
+ Description.builder()
+ .text("Read the store type of the paimon table.")
+ .linebreak()
+ .linebreak()
+ .text("Possible values:")
+ .linebreak()
+ .list(text("\"file\": Will read from the file store"))
+ .list(
+ text(
+ "\"log-system\":Will read from the file store, the user must configure log.system"))
Review Comment:
Ditto.
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #812: [core]Add streaming read from option (#778)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #812:
URL: https://github.com/apache/incubator-paimon/pull/812#discussion_r1159740807
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java:
##########
@@ -142,7 +144,10 @@ private ContinuousFileStoreSource buildContinuousFileSource() {
// TODO visit all options through CoreOptions
StartupMode startupMode = CoreOptions.startupMode(conf);
- if (logSourceProvider == null) {
+ StreamReadType streamReadType = CoreOptions.streamReadType(conf);
+
+ // no log system or
+ if (logSourceProvider == null || streamReadType == FILE) {
Review Comment:
no log system or log system is configured but streamReadType is FILE
--
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: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org