[FLINK-38866][flink-table-planner] Passing timestamp precision from window strategy to window operators to support non-compact BinaryRowData timestamps

[mattcuento]

What is the purpose of the change

BinaryRowData stores timestamps in a non-compact form when precision is > 3. Currently, window operator code hardcodes precision to 3. So, when reading back a non-compact timestamp, incorrect deserialization occurs resulting in incorrect windows.

Users are only exposed to this bug when using BinaryRowData with a descriptor that has precision > 3 and is running the job in batch mode.

In streaming mode, descriptors must be event time attributes, which must be a precision of 3 at maximum by default.

Brief change log

• Extract the timestamp precision from the window assigner in StreamExecWindowTableFunction to pass to UnalignedWindowTableFunctionOperator
• Extract the timestamp precision from the window assigner in CommonExecWindowTableFunction to pass to AlignedWindowTableFunctionOperator
• Extend WindowTableFunctionOperatorBase constructor to take in timestampPrecision as an int in a protected field
• Extend AlignedWindowTableFunctionOperator and UnalignedWindowTableFunctionOperator constructors to take in timestampPrecision int argument, use it in processElement for dynamically handling non-compact BinaryRowData

Verifying this change

Please make sure both new and modified tests in this PR follow the conventions for tests defined in our code quality guide.

This change added tests and can be verified as follows:

• Added an end-to-end Batch SQL test with non-compact/compact timestamp representations on a query that forces use of BinaryRowData

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not applicable)

[flinkbot]

CI report:

• 32ebe4a Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

I think it would be cleaner to keep the original method signature, and call the super(windowAssigner, rowtimeIndex,shiftTimeZone); The default then live in the WindowTableFunctionOperatorBase, rather than being specified in all the callers

The callers that need to pass the precision can use the new method.

[mattcuento]

I might be a bit confused on what you mean. Is the desire for a new constructor in WindowTableFunctionOperatorBase with a signature that passes timestampPrecision?

If so, the only two usages are for these two implementations, AlignedWindowTableFunctionOperator and UnalignedWindowTableFunctionOperator. Intended behavior here would be to always pass timestampPrecision through a concrete operator instance as opposed to defaulting to any given value

[mattcuento]

@flinkbot run azure