snuyanzin commented on code in PR #29003:
URL: https://github.com/apache/flink/pull/29003#discussion_r3971662306
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSort.java:
##########
@@ -100,12 +103,25 @@ public StreamExecSort(
@Override
protected Transformation<RowData> translateToPlanInternal(
PlannerBase planner, ExecNodeConfig config) {
+ ExecEdge inputEdge = getInputEdges().get(0);
+ RowType inputType = (RowType) inputEdge.getOutputType();
if
(!config.get(InternalConfigOptions.TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED)) {
- throw new TableException("Sort on a non-time-attribute field is
not supported.");
+ // Backstop for compiled plans loaded without passing through
StreamPhysicalSortRule.
+ if (sortSpec.getFieldSize() == 0) {
+ throw new TableException(
+ "Compiled plan contains a streaming sort without sort
keys.");
+ }
+ int firstSortField = sortSpec.getFieldIndices()[0];
+ String column = inputType.getFieldNames().get(firstSortField);
+ LogicalType type = inputType.getTypeAt(firstSortField);
+ if (LogicalTypeChecks.isTimeAttribute(type)
+ && !sortSpec.getFieldSpecs()[0].getIsAscendingOrder()) {
+ throw new TableException(
+
SortUtil.sortKeyTimeAttributeMustBeAscendingMessage(column));
+ }
+ throw new
TableException(SortUtil.sortKeyNotTimeAttributeMessage(column, type));
Review Comment:
is it correct condition?
for the case of descending order it will show nothing about the real issue
just something like this
```
Streaming ORDER BY requires the primary sort key to be a time attribute in
ascending order, but 'rowtime' is TIMESTAMP(3) *ROWTIME*
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
