[FLINK-39763][table] Allow inferring constants with IS NOT DISTINCT FROM

[snuyanzin]

What is the purpose of the change

The PRs enables Calcite's optimization for inferring constants https://issues.apache.org/jira/browse/CALCITE-5336

Brief change log

RexUtil

Verifying this change

Existing tests and updated tests

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)

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

[snuyanzin]

the constant pulled up

[snuyanzin]

here and for other joins (except for NestedLoop)
now Calcite is able to understand that a == 2 and e == 1
after that it is able to push down these values to corresponding table sources
as a result non nested loop join will fail as there is only true in join condition.

when all joins are available then NestedLoop will be selected, there is also e2e case in DecimalITCase in this PR

[flinkbot]

CI report:

• 20db663 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[jnh5y]

Do we need/want to add a test for IS NOT DISTINCT FROM or is that already covered sufficiently?

[jnh5y]

Why is this being changed?

[snuyanzin]

same reason as explained above #28256 (comment)

[jnh5y]

Seems like this comment needs to be updated?

[snuyanzin]

makes sense, updated

[snuyanzin]

Do we need/want to add a test for IS NOT DISTINCT FROM or is that already covered sufficiently?

if you mean IS NOT DISTINCT FROM in SQL the there are already some existing like

flink/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/SqlExpressionTest.scala

Lines 53 to 54 in d70e596

	testSqlApi("NULLIF(1,1) IS NOT DISTINCT FROM NULLIF(1,1)", "TRUE")
	testSqlApi("NULLIF(1,1) IS NOT DISTINCT FROM NULLIF(1,1)", "TRUE")

flink/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/join/WindowJoinTest.scala

Lines 1259 to 1288 in d70e596

	@Test
	def testJoinWithIsNotDistinctFrom(): Unit = {
	val sql =
	"""
	|SELECT L.*, R.*
	|FROM (
	| SELECT
	| a,
	| window_start,
	| window_end,
	| window_time,
	| count(*) as cnt,
	| count(distinct c) AS uv
	| FROM TABLE(TUMBLE(TABLE MyTable, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE))
	| GROUP BY a, window_start, window_end, window_time
	|) L
	|JOIN (
	| SELECT
	| a,
	| window_start,
	| window_end,
	| window_time,
	| count(*) as cnt,
	| count(distinct c) AS uv
	| FROM TABLE(TUMBLE(TABLE MyTable2, DESCRIPTOR(rowtime), INTERVAL '15' MINUTE))
	| GROUP BY a, window_start, window_end, window_time
	|) R
	|ON L.window_start = R.window_start AND L.window_end = R.window_end AND
	|L.a IS NOT DISTINCT FROM R.a
	""".stripMargin