You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "dawidwys (via GitHub)" <gi...@apache.org> on 2024/10/15 15:46:16 UTC
[PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
dawidwys opened a new pull request, #25526:
URL: https://github.com/apache/flink/pull/25526
## What is the purpose of the change
Fix string serialization of `OVER` window
## Verifying this change
added tests
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #25526:
URL: https://github.com/apache/flink/pull/25526#discussion_r1802680725
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
+ String order = operands.get(1).asSerializableString();
+ String rangeBounds =
+ CallSyntaxUtils.overRangeToSerializableString(
+ operands.get(2), operands.get(3));
+ if (operands.size() == 4) {
+ return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ } else {
+ return String.format(
+ "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ projection,
+ CallSyntaxUtils.asSerializableOperand(operands.get(4)),
+ order,
+ rangeBounds);
+ }
Review Comment:
Alright, then nevermind.
I just thought that Table API in batch-mode might support `OVER` w/o `ORDER BY` and that it would need to be covered here as well.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25526:
URL: https://github.com/apache/flink/pull/25526#issuecomment-2414407107
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "dd9063d44816500fde9ffa38e922998ad67473cd",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dd9063d44816500fde9ffa38e922998ad67473cd",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dd9063d44816500fde9ffa38e922998ad67473cd UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #25526:
URL: https://github.com/apache/flink/pull/25526#discussion_r1802563488
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
+ String order = operands.get(1).asSerializableString();
+ String rangeBounds =
+ CallSyntaxUtils.overRangeToSerializableString(
+ operands.get(2), operands.get(3));
+ if (operands.size() == 4) {
+ return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ } else {
+ return String.format(
+ "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ projection,
+ CallSyntaxUtils.asSerializableOperand(operands.get(4)),
+ order,
+ rangeBounds);
+ }
Review Comment:
This is too high level to test or ask the questions you asked 😅 If you'd like to check or fix sth in how `OVER` windows work in general we could create a separate issue.
This covers only the unparsing and it unparses all supported cases. This can be checked with https://github.com/apache/flink/blob/1986cfe1cb3606fc00fd13735f7ba901b806f56f/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java#L48
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "fhueske (via GitHub)" <gi...@apache.org>.
fhueske commented on code in PR #25526:
URL: https://github.com/apache/flink/pull/25526#discussion_r1801586512
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
Review Comment:
```suggestion
String partitioning = operands.get(0).asSerializableString();
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java:
##########
@@ -740,4 +740,89 @@ private static Instant dayOfSeconds(int second) {
},
"MySink")
.build();
+
+ static final TableTestProgram OVER_WINDOW_RANGE =
Review Comment:
Should we also test the `UNBOUNDED PRECEDING` case here?
It is handled differently than the bounded preceding window case.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java:
##########
@@ -740,4 +740,89 @@ private static Instant dayOfSeconds(int second) {
},
"MySink")
.build();
+
+ static final TableTestProgram OVER_WINDOW_RANGE =
+ TableTestProgram.of("over-window-range", "test over window with time range")
+ .setupTableSource(
+ SourceTestStep.newBuilder("data")
+ .addSchema(
+ "k string",
+ "v bigint",
+ "ts TIMESTAMP_LTZ(3)",
+ "WATERMARK for `ts` AS `ts`")
+ .producedBeforeRestore(
+ Row.of("Apple", 5L, dayOfSeconds(0)),
+ Row.of("Apple", 4L, dayOfSeconds(1)))
+ .producedAfterRestore(Row.of("Apple", 3L, dayOfSeconds(2)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("k string", "v bigint", "ts TIMESTAMP_LTZ(3)")
+ .consumedBeforeRestore(
+ Row.of("Apple", 5L, dayOfSeconds(0)),
+ Row.of("Apple", 4L, dayOfSeconds(1)))
+ .consumedAfterRestore(Row.of("Apple", 3L, dayOfSeconds(2)))
+ .build())
+ .runSql(
+ "SELECT `k`, (LAST_VALUE(`v`) OVER(PARTITION BY `k` ORDER BY `ts` RANGE BETWEEN INTERVAL '0 00:00:02.0' DAY TO SECOND(3) PRECEDING AND CURRENT ROW)) AS `_c1`, `ts` FROM (\n"
+ + " SELECT `k`, `v`, `ts` FROM `default_catalog`.`default_database`.`data`\n"
+ + ")")
+ .runTableApi(
+ tableEnvAccessor ->
+ tableEnvAccessor
+ .from("data")
+ .window(
+ Over.partitionBy($("k"))
+ .orderBy($("ts"))
+ .preceding(lit(2).second())
+ .as("w"))
+ .select(
+ $("k"),
+ $("v").lastValue().over($("w")),
+ $("ts")),
+ "sink")
+ .build();
+
+ static final TableTestProgram OVER_WINDOW_ROWS =
Review Comment:
Should we also test the `UNBOUNDED PRECEDING` case here?
It is handled differently than the bounded preceding window case.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java:
##########
@@ -740,4 +740,89 @@ private static Instant dayOfSeconds(int second) {
},
"MySink")
.build();
+
+ static final TableTestProgram OVER_WINDOW_RANGE =
Review Comment:
+ we should also test `OVER` without `PARTITION BY` (and possibly without `ORDER BY`)
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
+ String order = operands.get(1).asSerializableString();
+ String rangeBounds =
+ CallSyntaxUtils.overRangeToSerializableString(
+ operands.get(2), operands.get(3));
+ if (operands.size() == 4) {
+ return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ } else {
+ return String.format(
+ "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ projection,
+ CallSyntaxUtils.asSerializableOperand(operands.get(4)),
+ order,
+ rangeBounds);
+ }
Review Comment:
I think this logic might not be sufficient to capture the full scope of `OVER`
Neither `PARTITION BY` or `ORDER BY` clauses are required for `OVER` windows.
Without `PARTITION BY`, we cannot parallelize the operation. This is supported here.
However, in batch mode, also `ORDER BY` is optional. Without `ORDER BY`, we cannot specify a range so it is always the full partition. Some aggregation functions need `ORDER BY`, such as `LAST_VALUE()`, `ROW_NUMBER()`, etc.., but others like `SUM()` work fine without order by (on the full partition).
I'm not sure which variants of `OVER` we support in batch mode, but IMO it would be good to support all variants that are valid according to the SQL standard (if possible).
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #25526:
URL: https://github.com/apache/flink/pull/25526#discussion_r1802865897
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
+ String order = operands.get(1).asSerializableString();
+ String rangeBounds =
+ CallSyntaxUtils.overRangeToSerializableString(
+ operands.get(2), operands.get(3));
+ if (operands.size() == 4) {
+ return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ } else {
+ return String.format(
+ "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ projection,
+ CallSyntaxUtils.asSerializableOperand(operands.get(4)),
+ order,
+ rangeBounds);
+ }
Review Comment:
I created https://issues.apache.org/jira/browse/FLINK-33410
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #25526:
URL: https://github.com/apache/flink/pull/25526#discussion_r1802563488
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
+ String order = operands.get(1).asSerializableString();
+ String rangeBounds =
+ CallSyntaxUtils.overRangeToSerializableString(
+ operands.get(2), operands.get(3));
+ if (operands.size() == 4) {
+ return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ } else {
+ return String.format(
+ "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ projection,
+ CallSyntaxUtils.asSerializableOperand(operands.get(4)),
+ order,
+ rangeBounds);
+ }
Review Comment:
This is too high level to test or ask the questions you asked 😅 If you'd like to check or fix sth in how `OVER` windows work in general we could create a separate issue.
This covers only the unparsing and it unparses all supported cases. The expected signature can be checked with https://github.com/apache/flink/blob/1986cfe1cb3606fc00fd13735f7ba901b806f56f/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/OverTypeStrategy.java#L48
It expects:
1. time attribute at the second position
2. range arguments at 3 and 4
3. optional 5th argument containing partitioning key
4. a function as the first argument
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on code in PR #25526:
URL: https://github.com/apache/flink/pull/25526#discussion_r1802857026
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/SqlCallSyntax.java:
##########
@@ -316,4 +316,23 @@ private String doUnParse(
CallSyntaxUtils.asSerializableOperand(operands.get(2)));
}
};
+
+ SqlCallSyntax OVER =
+ ((sqlName, operands) -> {
+ String projection = operands.get(0).asSerializableString();
+ String order = operands.get(1).asSerializableString();
+ String rangeBounds =
+ CallSyntaxUtils.overRangeToSerializableString(
+ operands.get(2), operands.get(3));
+ if (operands.size() == 4) {
+ return String.format("%s OVER(ORDER BY %s %s)", projection, order, rangeBounds);
+ } else {
+ return String.format(
+ "%s OVER(PARTITION BY %s ORDER BY %s %s)",
+ projection,
+ CallSyntaxUtils.asSerializableOperand(operands.get(4)),
+ order,
+ rangeBounds);
+ }
Review Comment:
Generally, good catch. I just think it's a broader issue to make `OVER` work properly in batch as well. I'll create a ticket for that.
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys merged PR #25526:
URL: https://github.com/apache/flink/pull/25526
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36543] Fix string serialization for OVER windows [flink]
Posted by "dawidwys (via GitHub)" <gi...@apache.org>.
dawidwys commented on PR #25526:
URL: https://github.com/apache/flink/pull/25526#issuecomment-2414392759
cc @fhueske
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org