https://flink.apache.org/ logo
Join Slack
Powered by
Hi, I am using MATCH_RECOGNIZE for pattern matchin...
# troubleshooting
k
Hi, I am using MATCH_RECOGNIZE for pattern matching in events, I am trying to find average usage less than 60 withing 10 seconds interval, the issue here is that in result i am only seeing only 2 result (till the marked position in the data bellow) not sure why further data is not processed or there is some problem with my query.
Copy code
CREATE TABLE CPU_IN_TIMEBASED (
	hostname STRING,
	cpu STRING,
	usage DOUBLE,
	occurred_at BIGINT,
	time_ltz AS TO_TIMESTAMP_LTZ(occurred_at, 3),
	WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECONDS
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = 'localhost:9092',
  'topic' = 'cpu_load_timebased',
  'value.format' = 'json',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'properties.auto.offset.reset' = 'earliest'
);

SELECT * FROM CPU_IN_TIMEBASED
MATCH_RECOGNIZE (
	PARTITION BY hostname
	ORDER BY time_ltz
	MEASURES
		FIRST(A.time_ltz) as start_tstamp,
		LAST(A.time_ltz) as end_tstamp,
		AVG(A.usage) as avgUsage,
		A.cpu as brain,
		A.occurred_at as ts
  AFTER MATCH SKIP PAST LAST ROW
	PATTERN (A+ B) WITHIN INTERVAL '5' SECOND
	DEFINE
		A AS AVG(A.usage) < 60
) MR;
Data:
Copy code
{"hostname":"one","cpu":"cpu1","usage":95,"occurred_at":1691645762224}
{"hostname":"one","cpu":"cpu1","usage":35,"occurred_at":1691645764078}
{"hostname":"one","cpu":"cpu1","usage":63,"occurred_at":1691645765122}
{"hostname":"one","cpu":"cpu1","usage":24,"occurred_at":1691645766158}
{"hostname":"one","cpu":"cpu1","usage":86,"occurred_at":1691645767188}
{"hostname":"one","cpu":"cpu1","usage":82,"occurred_at":1691645768286}
{"hostname":"one","cpu":"cpu1","usage":73,"occurred_at":1691645769315}
{"hostname":"one","cpu":"cpu1","usage":23,"occurred_at":1691645770335}
{"hostname":"one","cpu":"cpu1","usage":37,"occurred_at":1691645771365} <- result till here
{"hostname":"one","cpu":"cpu1","usage":84,"occurred_at":1691645772383}
{"hostname":"one","cpu":"cpu1","usage":39,"occurred_at":1691645773408}
{"hostname":"one","cpu":"cpu1","usage":18,"occurred_at":1691645774428}
{"hostname":"one","cpu":"cpu1","usage":77,"occurred_at":1691645775459}
{"hostname":"one","cpu":"cpu1","usage":91,"occurred_at":1691645776478}
{"hostname":"one","cpu":"cpu1","usage":35,"occurred_at":1691645777498}
{"hostname":"one","cpu":"cpu1","usage":28,"occurred_at":1691645778532}
{"hostname":"one","cpu":"cpu1","usage":41,"occurred_at":1691645779674}
{"hostname":"one","cpu":"cpu1","usage":50,"occurred_at":1691645780703}
{"hostname":"one","cpu":"cpu1","usage":81,"occurred_at":1691645781723}
{"hostname":"one","cpu":"cpu1","usage":20,"occurred_at":1691645782755}
{"hostname":"one","cpu":"cpu1","usage":35,"occurred_at":1691645783774}
{"hostname":"one","cpu":"cpu1","usage":43,"occurred_at":1691645784794}
{"hostname":"one","cpu":"cpu1","usage":22,"occurred_at":1691645785812}
{"hostname":"one","cpu":"cpu1","usage":37,"occurred_at":1691645786831}
{"hostname":"one","cpu":"cpu1","usage":64,"occurred_at":1691645787861}
d
The results are lagging 10 seconds behind the latest event because you have a 5-second-long watermarking delay, and a 5-second-long pattern matching interval.
k
I tried by removing the 5s lag from the water mark but I don't get how Flink is matching the events I guess that if I don't provide Definition for 
B
then it would match any row write? if that's the case then the result is not what it is supposed to be as, I have 
AFTER MATCH SKIP PAST LAST ROW
in my query that means the 2nd match should start from the 6th event (same for the further events) but the in the result I can observe that its not the case. can you correct me if I have got it wrong?
Copy code
SELECT * FROM CPU_IN_TIMEBASED
   MATCH_RECOGNIZE (
   ORDER BY time_ltz
   MEASURES
     FIRST(A.time_ltz) as start_tstamp,
     AVG(A.usage) as avgUsage,
     AVG(B.usage) as avgBUsage,
     COUNT(A.usage) as cntA,
     COUNT(B.usage) as cntB,
     A.occurred_at as ts,
     B.occurred_at as tsB
  AFTER MATCH SKIP PAST LAST ROW
  PATTERN (A+ B) WITHIN INTERVAL '5' SECOND
  DEFINE
    A AS A.hostname = 'one'
) MR;
Copy code
1. {"hostname":"one","cpu":"cpu1","usage":39,"occurred_at":1691749275211}    -
2. {"hostname":"one","cpu":"cpu1","usage":40,"occurred_at":1691749276211}     |
3. {"hostname":"one","cpu":"cpu1","usage":45,"occurred_at":1691749277211}     |  -> 1st match
4. {"hostname":"one","cpu":"cpu1","usage":32,"occurred_at":1691749278211}     |
5. {"hostname":"two","cpu":"cpu1","usage":31,"occurred_at":1691749279211}    -
6. {"hostname":"one","cpu":"cpu1","usage":50,"occurred_at":1691749280211}    skipped?
7. {"hostname":"one","cpu":"cpu1","usage":62,"occurred_at":1691749281211}    -
8. {"hostname":"one","cpu":"cpu1","usage":41,"occurred_at":1691749282211}     |
9. {"hostname":"one","cpu":"cpu1","usage":56,"occurred_at":1691749283211}     |  -> 2nd match
10. {"hostname":"one","cpu":"cpu1","usage":44,"occurred_at":1691749284211}    |
11. {"hostname":"two","cpu":"cpu1","usage":34,"occurred_at":1691749285211}   -
12. {"hostname":"one","cpu":"cpu1","usage":61,"occurred_at":1691749286211}   skipped?
13. {"hostname":"one","cpu":"cpu1","usage":53,"occurred_at":1691749287211}   skipped?
14. {"hostname":"one","cpu":"cpu1","usage":50,"occurred_at":1691749288211}   -
15. {"hostname":"one","cpu":"cpu1","usage":42,"occurred_at":1691749289211}    |
16. {"hostname":"one","cpu":"cpu1","usage":46,"occurred_at":1691749290211}    |  -> 3rd match
17. {"hostname":"one","cpu":"cpu1","usage":56,"occurred_at":1691749291211}    |
18. {"hostname":"two","cpu":"cpu1","usage":35,"occurred_at":1691749292211}   -
@David Anderson correct me if I am wrong, the above behavior is because of no match found and Flink will try to match new events by skipping one row similar as 
AFTER MATCH SKIP TO NEXT ROW?
d
I guess that if I don't provide Definition for 
B
then it would match any row write?
Yes, that's right. My guess is that the behavior you're seeing has to do with the definition of "within 5 seconds". You are expecting a window starting with event 6 to include events 6-11, which spans [1691749280211, 1691749285211]. My guess is that the implementation would only allow for [1691749280211, 1691749285211) -- I suspect the interval between events 6 and 11 is one millisecond too long.
k
okay got it, thanks 🙌
Hi, I am using Flink CEP with SQL for processing events, the schema is determined by the JSON event, while trying to execute the given SQL it gives bellow exception can. we have defined DTO’s for the log, does any one know what am I doing wrong. as basic query like 
SELECT message.event.size FROM Log
works just fine.
Copy code
{
  "message": {
    "@timestamp": "2023-08-02T12:12:16Z",
    "customFields": "...",
    "cloud": {
      "region": "us-east-1"
    },
    "ecs": {
      "version": "8.0.0"
    },
    "event": {
      "action": "GetBucketAcl",
      "id": "bb7ee0ed-f273-4da8-a75d-53e10e7d1515",
      "kind": "event",
      "outcome": "success",
      "provider": "<http://s3.amazonaws.com|s3.amazonaws.com>",
      "type": "info",
      "size": 10
    },
    "source": {
      "address": "<http://cloudtrail.amazonaws.com|cloudtrail.amazonaws.com>"
    }
  },
  "id": "pxX_35all1pkteGtxXxOi",
  "timestamp": "2023-08-02 12:17:31.526356",
  "integration": "cloudtrail",
  "configuration": "conf-1"
}
Copy code
SELECT * FROM Log
MATCH_RECOGNIZE (
	PARTITION BY integration
	ORDER BY time_ltz
	MEASURES
		FIRST(A.time_ltz) AS start_tstamp,
		LAST(A.time_ltz) AS end_tstamp,
		SUM(message.event.size) AS size,
		A.id AS aID,
		LAST(B.id) as bID,
		C.id AS cID
	AFTER MATCH SKIP PAST LAST ROW
	PATTERN (A B+ C) WITHIN INTERVAL '5' SECOND
	DEFINE
		A AS A.message.event.action = 'login',
		B AS SUM(message.event.size) > 30
) T;
Copy code
(
  `message` *org.abs.security.CloudTrailDto.Message<`cloud` *org.abs.security.CloudTrailDto.Cloud<`region` STRING>*, `customField` STRING, `ecs` *org.abs.security.CloudTrailDto.Ecs<`version` STRING>*, `event` *org.abs.security.CloudTrailDto.Event<`action` STRING, `id` STRING, `kind` STRING, `outcome` STRING, `provider` STRING, `type` STRING, `size` BIGINT NOT NULL>*, `source` *org.abs.security.CloudTrailDto.Source<`address` STRING>*, `timestamp` STRING>*,
  `id` STRING,
  `timestamp` STRING,
  `integration` STRING,
  `configuration` STRING,
  `time_ltz` TIMESTAMP(3)
)
Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. From line 15, column 22 to line 15, column 26: Table 'event.message.event' not found
	at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:187)
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:113)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:281)
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:106)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:738)
	at org.abs.security.StaticRuleEvaluator.main(StaticRuleEvaluator.java:115)
Caused by: org.apache.calcite.runtime.CalciteContextException: From line 15, column 22 to line 15, column 26: Table 'event.message.event' not found
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)
	at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:505)
	at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:932)
	at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:917)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:5163)
	at org.apache.calcite.sql.validate.DelegatingScope.fullyQualify(DelegatingScope.java:344)
	at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:6241)
	at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:6203)
	at org.apache.calcite.sql.SqlIdentifier.accept(SqlIdentifier.java:324)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveTypeImpl(SqlValidatorImpl.java:1861)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveType(SqlValidatorImpl.java:1852)
	at org.apache.calcite.sql.SqlOperator.constructArgTypeList(SqlOperator.java:680)
	at org.apache.calcite.sql.SqlFunction.deriveType(SqlFunction.java:249)
	at org.apache.calcite.sql.SqlFunction.deriveType(SqlFunction.java:231)
	at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:6218)
	at org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:6203)
	at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:161)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveTypeImpl(SqlValidatorImpl.java:1861)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.deriveType(SqlValidatorImpl.java:1852)
	at org.apache.calcite.sql.type.SqlTypeUtil.deriveType(SqlTypeUtil.java:178)
	at org.apache.calcite.sql.type.InferTypes.lambda$static$0(InferTypes.java:47)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.inferUnknownTypes(SqlValidatorImpl.java:2042)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.inferUnknownTypes(SqlValidatorImpl.java:2030)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateDefinitions(SqlValidatorImpl.java:5487)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateMatchRecognize(SqlValidatorImpl.java:5365)
	at org.apache.calcite.sql.validate.MatchRecognizeNamespace.validateImpl(MatchRecognizeNamespace.java:42)
	at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:89)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1042)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1017)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateFrom(SqlValidatorImpl.java:3287)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelect(SqlValidatorImpl.java:3558)
	at org.apache.calcite.sql.validate.SelectNamespace.validateImpl(SelectNamespace.java:64)
	at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:89)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1042)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1017)
	at org.apache.calcite.sql.SqlSelect.validate(SqlSelect.java:247)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:992)
	at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:741)
	at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:183)
	... 5 more
Caused by: org.apache.calcite.sql.validate.SqlValidatorException: Table 'event.message.event' not found
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.base/java.lang.reflect.Constructor.newInstance(Constructor.java:490)
	at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:505)
	at org.apache.calcite.runtime.Resources$ExInst.ex(Resources.java:599)
	... 42 more

Process finished with exit code 1
d
Since the pattern only includes one A, this part
Copy code
FIRST(A.time_ltz) AS start_tstamp,
LAST(A.time_ltz) AS end_tstamp,
doesn't seem quite right, but I don't believe that's the problem.
k
yes that completely unnecessary, but the question here is that can I use this part will this work because it is not working for me (
A.message.event.action,
while accessing inner objects), it just works fine when not in MATCH_RECOGNIZE?
Copy code
DEFINE
		A AS A.message.event.action = 'login',
		B AS SUM(message.event.size) > 30
) T;
I am trying to access inner object
d
The A in 
A.message.event.action
seems unnecessary, but should be okay. Maybe quoting it, as in
Copy code
A AS `message.event.action` = 'login',
B AS SUM(`message.event.size`) > 30
would help. (I don’t know what’s wrong; grasping at straws.)
k
I tried quoting but then the whole thing is considered as a single column.
d
Ahh, what about this, then:
Copy code
A AS `message`.`event`.`action` = 'login',
B AS SUM(`message`.`event`.`size`) > 30
k
I could try that, ill let you know if itworks
Nope, that doesn't work, getting following ereror
Copy code
Exception in thread "main" java.lang.RuntimeException: while converting FINAL SUM(`*`.`message`.`size`)
	at org.apache.calcite.sql2rel.ReflectiveConvertletTable.lambda$registerNodeTypeMethod$1(ReflectiveConvertletTable.java:98)
	at org.apache.calcite.sql2rel.SqlNodeToRexConverterImpl.convertCall(SqlNodeToRexConverterImpl.java:62)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:5318)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:4529)
	at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:161)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.convertExpression(SqlToRelConverter.java:5107)
	at org.apache.calcite.sql2rel.StandardConvertletTable.lambda$new$9(StandardConvertletTable.java:209)
	at org.apache.calcite.sql2rel.SqlNodeToRexConverterImpl.convertCall(SqlNodeToRexConverterImpl.java:62)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:5318)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:4529)
	at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:161)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.convertExpression(SqlToRelConverter.java:5107)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertMatchRecognize(SqlToRelConverter.java:2454)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2181)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2092)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:700)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:686)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3589)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:599)
	at <http://org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org|org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org>$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:216)
	at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:192)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:1580)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:1285)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertValidatedSqlNode(SqlToOperationConverter.java:397)
	at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:282)
	at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:106)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:738)
	at org.abs.security.StaticRuleEvaluator.main(StaticRuleEvaluator.java:136)
Caused by: java.lang.reflect.InvocationTargetException
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.calcite.sql2rel.ReflectiveConvertletTable.lambda$registerNodeTypeMethod$1(ReflectiveConvertletTable.java:93)
	... 27 more
Caused by: java.lang.RuntimeException: while converting SUM(`*`.`message`.`size`)
	at org.apache.calcite.sql2rel.ReflectiveConvertletTable.lambda$registerOpTypeMethod$3(ReflectiveConvertletTable.java:145)
	at org.apache.calcite.sql2rel.SqlNodeToRexConverterImpl.convertCall(SqlNodeToRexConverterImpl.java:62)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:5318)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:4529)
	at org.apache.calcite.sql.SqlCall.accept(SqlCall.java:161)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.convertExpression(SqlToRelConverter.java:5107)
	at org.apache.calcite.sql2rel.StandardConvertletTable.convertOperands(StandardConvertletTable.java:968)
	at org.apache.calcite.sql2rel.StandardConvertletTable.convertOperands(StandardConvertletTable.java:960)
	at org.apache.calcite.sql2rel.StandardConvertletTable.convertCall(StandardConvertletTable.java:923)
	... 32 more
Caused by: java.lang.reflect.InvocationTargetException
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at org.apache.calcite.sql2rel.ReflectiveConvertletTable.lambda$registerOpTypeMethod$3(ReflectiveConvertletTable.java:140)
	... 40 more
Caused by: java.lang.NullPointerException: no expression found for {id: message.size, prefix: 2}
	at java.base/java.util.Objects.requireNonNull(Objects.java:347)
	at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:4167)
	at org.apache.calcite.sql2rel.SqlToRelConverter.access$2500(SqlToRelConverter.java:230)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:5329)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.visit(SqlToRelConverter.java:4529)
	at org.apache.calcite.sql.SqlIdentifier.accept(SqlIdentifier.java:324)
	at org.apache.calcite.sql2rel.SqlToRelConverter$Blackboard.convertExpression(SqlToRelConverter.java:5107)
	at org.apache.calcite.sql2rel.StandardConvertletTable.convertOperands(StandardConvertletTable.java:968)
	at org.apache.calcite.sql2rel.StandardConvertletTable.convertOperands(StandardConvertletTable.java:960)
	at org.apache.calcite.sql2rel.StandardConvertletTable.convertAggregateFunction(StandardConvertletTable.java:810)
	... 45 more
d
That looks like progress (it’s a different error). Can you try
Copy code
A AS `message`.`event`.`action` = 'login',
B AS SUM(B.`message`.`event`.`size`) > 30
k
I tried with this but got the following error
Copy code
A AS A.`message`.`action` = 'login',
B AS B.`message`.`action` = 'data' AND SUM(B.`message`.`size`) >= 20
Copy code
Exception in thread "main" org.apache.flink.table.api.TableException: No pattern variable specified in aggregate
	at org.apache.flink.table.planner.codegen.MatchCodeGenerator.$anonfun$visitCall$1(MatchCodeGenerator.scala:360)
	at scala.Option.getOrElse(Option.scala:121)
	at org.apache.flink.table.planner.codegen.MatchCodeGenerator.visitCall(MatchCodeGenerator.scala:360)
	at org.apache.flink.table.planner.codegen.MatchCodeGenerator.visitCall(MatchCodeGenerator.scala:84)
	at org.apache.calcite.rex.RexCall.accept(RexCall.java:189)
	at org.apache.flink.table.planner.codegen.ExprCodeGenerator.$anonfun$visitCall$1(ExprCodeGenerator.scala:487)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
	at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:58)
	at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:51)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
	at scala.collection.TraversableLike.map(TraversableLike.scala:233)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
	at org.apache.flink.table.planner.codegen.ExprCodeGenerator.visitCall(ExprCodeGenerator.scala:478)
	at org.apache.flink.table.planner.codegen.MatchCodeGenerator.visitCall(MatchCodeGenerator.scala:388)
	at org.apache.flink.table.planner.codegen.MatchCodeGenerator.visitCall(MatchCodeGenerator.scala:84)
	at org.apache.calcite.rex.RexCall.accept(RexCall.java:189)
	at org.apache.flink.table.planner.codegen.ExprCodeGenerator.$anonfun$visitCall$1(ExprCodeGenerator.scala:487)
A.message.action
worked fine but it didn't worked with Aggregate function and 
A.action
works with or without Aggregate function. To me it seams like Aggregate functions does not support nested/inner object and without Aggregate function it only works to 2 level deep?
d
I think this could be a bug. Can you create a JIRA ticket: https://issues.apache.org/jira/projects/FLINK/issues/FLINK-32726?filter=allopenissues
k
2 Views
Open in Slack
PreviousNext
Powered by