[Bug]: Beam SQL cannot work with array[DATETIME] field

[vampire-hunter-lincoln]

What happened?

Hi Beam community,

I found a Beam SQL issue recently when using Beam Java SDK together with Beam SQL.

In my project, a beam row has such a schema for a timestamp array field: org.apache.beam.sdk.schemas.Schema.FieldType.array(org.apache.beam.sdk.schemas.Schema.FieldType.DATETIME), then I apply a simple Beam SQL query SELECT * FROM PCOLLECTION on the beam rows. The field value is a list of Joda Instant.

Finally, I got the following error message in Dataflow Runner:

Caused by: java.lang.ClassCastException: class org.joda.time.Instant cannot be cast to class org.joda.time.DateTime (org.joda.time.Instant and org.joda.time.DateTime are in unnamed module of loader 'app')
SC$1.value(Unknown Source)
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel$WrappedList.get(BeamCalcRel.java:862)
java.base/java.util.AbstractList$Itr.next(AbstractList.java:371)
java.base/java.util.Iterator.forEachRemaining(Iterator.java:133)
java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801)
java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484)
java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474)
java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:913)
java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:578)
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel.toBeamList(BeamCalcRel.java:454)
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel.toBeamObject(BeamCalcRel.java:399)
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel.toBeamRow(BeamCalcRel.java:475)
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel.access$700(BeamCalcRel.java:125)
org.apache.beam.sdk.extensions.sql.impl.rel.BeamCalcRel$CalcFn.processElement(BeamCalcRel.java:318)
....

Direct Java runner doesn't throw this exception when running the same code.

Beam Java SDK: 2.64.0 and 2.66.0

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner