You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "swapna267 (via GitHub)" <gi...@apache.org> on 2025/02/07 00:38:20 UTC
[PR] support source watermark for flink sql windows [iceberg]
swapna267 opened a new pull request, #12191:
URL: https://github.com/apache/iceberg/pull/12191
Iceberg Source to support Source Watermark, so it can be used in Flink WINDOW functions. https://github.com/apache/flink/blob/release-1.18/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/source/abilities/SupportsSourceWatermark.java enables Flink to rely on the watermark strategy provided by the ScanTableSource itself.
```
CREATE TABLE table_wm (
eventTS AS CAST(t1 AS TIMESTAMP(3)),
WATERMARK FOR eventTS AS SOURCE_WATERMARK()
) WITH (
'watermark-column'='t1'
) LIKE iceberg_catalog.db.table;
```
**Reference:**
https://github.com/apache/iceberg/issues/10219
https://github.com/apache/iceberg/pull/9346
Previous discussion in PR, https://github.com/apache/iceberg/pull/12116 . Split into separate PR for easy review.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "swapna267 (via GitHub)" <gi...@apache.org>.
swapna267 commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1990246723
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -175,6 +178,18 @@ public Result applyFilters(List<ResolvedExpression> flinkFilters) {
return Result.of(acceptedFilters, flinkFilters);
}
+ @Override
+ public void applySourceWatermark() {
+ if (!readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE)) {
Review Comment:
Thought it would be better to throw UnsupportedOperation Exception instead of IllegalArgument or IllegalStateException. If either of them is fine with current error message , I can move to Preconditions.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support source watermark for flink sql windows [iceberg]
Posted by "swapna267 (via GitHub)" <gi...@apache.org>.
swapna267 commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1947355037
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/SqlBase.java:
##########
@@ -86,7 +86,7 @@ protected void dropDatabase(String database, boolean ifExists) {
sql("DROP DATABASE %s %s", ifExists ? "IF EXISTS" : "", database);
}
- protected static String toWithClause(Map<String, String> props) {
+ public static String toWithClause(Map<String, String> props) {
Review Comment:
This is currently being across Tests under org.apache.iceberg.flink. Wanted to use for test under org.apache.iceberg.flink.source. org.apache.iceberg.flink.source.SqlHelpers can be another option, where this can be done.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "swapna267 (via GitHub)" <gi...@apache.org>.
swapna267 commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1990251133
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -53,7 +55,11 @@ public class TestIcebergSourceSql extends TestSqlBase {
@BeforeEach
@Override
public void before() throws IOException {
- TableEnvironment tableEnvironment = getTableEnv();
+ setUpTableEnv(getTableEnv());
Review Comment:
Yes Source watermarks are being called on Source only in streaming env.
For iceberg source even in Batch mode, it would still be important to configure a watermark column, so the splits are sorted. But applySourceWatermark is not being called in batch env.
More details here,
https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/datastream/execution_mode/#event-time--watermarks
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1993384864
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -53,7 +55,11 @@ public class TestIcebergSourceSql extends TestSqlBase {
@BeforeEach
@Override
public void before() throws IOException {
- TableEnvironment tableEnvironment = getTableEnv();
+ setUpTableEnv(getTableEnv());
Review Comment:
Thx for clarifying
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary merged PR #12191:
URL: https://github.com/apache/iceberg/pull/12191
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "swapna267 (via GitHub)" <gi...@apache.org>.
swapna267 commented on PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#issuecomment-2737924089
Thanks all. Will submit backport PR soon.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1946012368
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/SqlBase.java:
##########
@@ -86,7 +86,7 @@ protected void dropDatabase(String database, boolean ifExists) {
sql("DROP DATABASE %s %s", ifExists ? "IF EXISTS" : "", database);
}
- protected static String toWithClause(Map<String, String> props) {
+ public static String toWithClause(Map<String, String> props) {
Review Comment:
Why is this change? Did this become a utility method which is shared between tests? Can we find a better place for it then?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1949097510
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -162,4 +173,74 @@ public void testWatermarkOptionsDescending() throws Exception {
expected,
SCHEMA_TS);
}
+
+ @Test
+ public void testWatermarkInvalidConfig() {
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA_TS);
+
+ String flinkTable = "`default_catalog`.`default_database`.flink_table";
+ try {
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "CREATE TABLE %s "
+ + "(t1 TIMESTAMP(6), "
+ + "t2 BIGINT,"
+ + "eventTS AS CAST(t1 AS TIMESTAMP(3)), "
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) WITH %s",
+ flinkTable,
+ SqlBase.toWithClause(getConnectorOptions()));
+
+ assertThatThrownBy(
+ () -> SqlHelpers.sql(getStreamingTableEnv(), "SELECT * FROM %s", flinkTable))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("watermark-column needs to be configured to use source watermark.");
+ } finally {
+ SqlHelpers.sql(getStreamingTableEnv(), "DROP TABLE IF EXISTS %s", flinkTable);
Review Comment:
In `TestSqlBase` we have several tests where we create tables, and we expect that they are removed because of the `@TempDir` is removed. So we can simply just do the same here.
WDYT?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1946014902
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -162,4 +173,74 @@ public void testWatermarkOptionsDescending() throws Exception {
expected,
SCHEMA_TS);
}
+
+ @Test
+ public void testWatermarkInvalidConfig() {
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA_TS);
+
+ String flinkTable = "`default_catalog`.`default_database`.flink_table";
+ try {
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "CREATE TABLE %s "
+ + "(t1 TIMESTAMP(6), "
+ + "t2 BIGINT,"
+ + "eventTS AS CAST(t1 AS TIMESTAMP(3)), "
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) WITH %s",
+ flinkTable,
+ SqlBase.toWithClause(getConnectorOptions()));
+
+ assertThatThrownBy(
+ () -> SqlHelpers.sql(getStreamingTableEnv(), "SELECT * FROM %s", flinkTable))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("watermark-column needs to be configured to use source watermark.");
+ } finally {
+ SqlHelpers.sql(getStreamingTableEnv(), "DROP TABLE IF EXISTS %s", flinkTable);
Review Comment:
Why is this try finally? Can we just do a cleanup in an after method? That would make the tests easier to read
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1993382151
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -175,6 +178,18 @@ public Result applyFilters(List<ResolvedExpression> flinkFilters) {
return Result.of(acceptedFilters, flinkFilters);
}
+ @Override
+ public void applySourceWatermark() {
+ if (!readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE)) {
Review Comment:
I would suggest to move to `Preconditions.checkArgument` - we usually stick to that for this kind of checks
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1949093445
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/SqlBase.java:
##########
@@ -86,7 +86,7 @@ protected void dropDatabase(String database, boolean ifExists) {
sql("DROP DATABASE %s %s", ifExists ? "IF EXISTS" : "", database);
}
- protected static String toWithClause(Map<String, String> props) {
+ public static String toWithClause(Map<String, String> props) {
Review Comment:
Sounds good
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support source watermark for flink sql windows [iceberg]
Posted by "swapna267 (via GitHub)" <gi...@apache.org>.
swapna267 commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1947376035
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -162,4 +173,74 @@ public void testWatermarkOptionsDescending() throws Exception {
expected,
SCHEMA_TS);
}
+
+ @Test
+ public void testWatermarkInvalidConfig() {
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA_TS);
+
+ String flinkTable = "`default_catalog`.`default_database`.flink_table";
+ try {
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "CREATE TABLE %s "
+ + "(t1 TIMESTAMP(6), "
+ + "t2 BIGINT,"
+ + "eventTS AS CAST(t1 AS TIMESTAMP(3)), "
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) WITH %s",
+ flinkTable,
+ SqlBase.toWithClause(getConnectorOptions()));
+
+ assertThatThrownBy(
+ () -> SqlHelpers.sql(getStreamingTableEnv(), "SELECT * FROM %s", flinkTable))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("watermark-column needs to be configured to use source watermark.");
+ } finally {
+ SqlHelpers.sql(getStreamingTableEnv(), "DROP TABLE IF EXISTS %s", flinkTable);
Review Comment:
Flink dynamic table is created only in specific tests in this class, so was handling here. But can do in after to improve readability.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1988454071
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestSqlBase.java:
##########
@@ -75,6 +77,18 @@ protected TableEnvironment getTableEnv() {
return tEnv;
}
+ protected TableEnvironment getStreamingTableEnv() {
+ if (streamingTEnv == null) {
+ synchronized (this) {
+ if (streamingTEnv == null) {
+ this.streamingTEnv =
+ TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());
+ }
+ }
+ }
Review Comment:
Nit: newline
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] support source watermark for flink sql windows [iceberg]
Posted by "swapna267 (via GitHub)" <gi...@apache.org>.
swapna267 commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1947348896
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -162,4 +173,74 @@ public void testWatermarkOptionsDescending() throws Exception {
expected,
SCHEMA_TS);
}
+
+ @Test
+ public void testWatermarkInvalidConfig() {
+ CATALOG_EXTENSION.catalog().createTable(TestFixtures.TABLE_IDENTIFIER, SCHEMA_TS);
+
+ String flinkTable = "`default_catalog`.`default_database`.flink_table";
+ try {
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "CREATE TABLE %s "
+ + "(t1 TIMESTAMP(6), "
+ + "t2 BIGINT,"
+ + "eventTS AS CAST(t1 AS TIMESTAMP(3)), "
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) WITH %s",
+ flinkTable,
+ SqlBase.toWithClause(getConnectorOptions()));
+
+ assertThatThrownBy(
+ () -> SqlHelpers.sql(getStreamingTableEnv(), "SELECT * FROM %s", flinkTable))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessage("watermark-column needs to be configured to use source watermark.");
+ } finally {
+ SqlHelpers.sql(getStreamingTableEnv(), "DROP TABLE IF EXISTS %s", flinkTable);
+ }
+ }
+
+ @Test
+ public void testWatermarkValidConfig() throws Exception {
+ List<Record> expected = generateExpectedRecords(true);
+
+ Map<String, String> connectorOptions = Maps.newHashMap(getConnectorOptions());
+ connectorOptions.put("watermark-column", "t1");
+
+ String flinkTable = "`default_catalog`.`default_database`.flink_table";
+
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "CREATE TABLE %s "
+ + "(t1 TIMESTAMP(6), "
+ + "t2 BIGINT,"
+ + "eventTS AS CAST(t1 AS TIMESTAMP(3)), "
+ + "WATERMARK FOR eventTS AS SOURCE_WATERMARK()) WITH %s",
+ flinkTable,
+ SqlBase.toWithClause(connectorOptions));
+
+ TestHelpers.assertRecordsWithOrder(
+ SqlHelpers.sql(
+ getStreamingTableEnv(),
+ "SELECT t1, t2 FROM TABLE(TUMBLE(TABLE %s, DESCRIPTOR(eventTS), INTERVAL '1' SECOND))",
+ flinkTable),
+ expected,
+ SCHEMA_TS);
+ }
+
+ @NotNull
+ private static Map<String, String> getConnectorOptions() {
Review Comment:
yes , that gives us an option to do create table LIKE and avoid connector properties.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#issuecomment-2737817710
Merged to main
Thanks for the PR @swapna267!
Thanks for the review @mxm and @stevenzwu!
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1988457157
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/source/TestIcebergSourceSql.java:
##########
@@ -53,7 +55,11 @@ public class TestIcebergSourceSql extends TestSqlBase {
@BeforeEach
@Override
public void before() throws IOException {
- TableEnvironment tableEnvironment = getTableEnv();
+ setUpTableEnv(getTableEnv());
Review Comment:
Could you please help me understand why the streaming env is required? Is Flink watermark strategy restricted to streaming?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#discussion_r1988451569
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -175,6 +178,18 @@ public Result applyFilters(List<ResolvedExpression> flinkFilters) {
return Result.of(acceptedFilters, flinkFilters);
}
+ @Override
+ public void applySourceWatermark() {
+ if (!readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE)) {
Review Comment:
Why is this not using `Preconditions` for the check?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support source watermark for flink sql windows [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #12191:
URL: https://github.com/apache/iceberg/pull/12191#issuecomment-2644565965
Nit: Renamed the PR to match the general patterns
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org