[FLINK-38234][table-planner] Enable NDU strategy for semantic tests by default

This closes #26899.

Author: gustavodemorais

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java

@@ -148,6 +148,7 @@ private static TableEnvironment setupEnv(TableTestProgram program) {
     @Override
     public EnumSet<TestKind> supportedSetupSteps() {
         return EnumSet.of(
+                TestKind.CONFIG,
                 TestKind.SQL,
                 TestKind.FUNCTION,
                 TestKind.SOURCE_WITH_DATA,

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java

@@ -19,6 +19,7 @@
 package org.apache.flink.table.api;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
 import org.apache.flink.table.functions.ScalarFunction;
 import org.apache.flink.table.operations.QueryOperation;
 import org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ChainedReceivingFunction;
@@ -1026,6 +1027,11 @@ private static Instant dayOfSeconds(int second) {
 
     public static final TableTestProgram ROW_SEMANTIC_TABLE_PTF =
             TableTestProgram.of("process-row-table-api", "table with row semantics")
+                    // TODO [FLINK-38233]: Remove this config when PTF support in
+                    //  StreamNonDeterministicUpdatePlanVisitor is added.
+                    .setupConfig(
+                            OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
+                            OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE)
                     .setupTemporarySystemFunction("f", RowSemanticTableFunction.class)
                     .setupSql(BASIC_VALUES)
                     .setupTableSink(
@@ -1051,6 +1057,11 @@ private static Instant dayOfSeconds(int second) {
 
     static final TableTestProgram SET_SEMANTIC_TABLE_PTF =
             TableTestProgram.of("partitioned-ptf", "verifies SQL serialization")
+                    // TODO [FLINK-38233]: Remove this config when PTF support in
+                    //  StreamNonDeterministicUpdatePlanVisitor is added.
+                    .setupConfig(
+                            OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
+                            OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE)
                     .setupTemporarySystemFunction("f1", ChainedSendingFunction.class)
                     .setupTemporarySystemFunction("f2", ChainedReceivingFunction.class)
                     .setupTableSource(TIMED_SOURCE)

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java

@@ -18,6 +18,8 @@
 
 package org.apache.flink.table.planner.plan.nodes.exec.stream;
 
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
 import org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
 import org.apache.flink.table.test.program.TableTestProgram;
 
@@ -26,6 +28,16 @@
 /** Semantic tests for {@link StreamExecProcessTableFunction}. */
 public class ProcessTableFunctionSemanticTests extends SemanticTestBase {
 
+    // TODO [FLINK-38233]: Remove this override when PTF support in
+    //  StreamNonDeterministicUpdatePlanVisitor is added.
+    @Override
+    protected void applyDefaultEnvironmentOptions(TableConfig config) {
+        super.applyDefaultEnvironmentOptions(config);
+        config.set(
+                OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
+                OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE);
+    }
+
     @Override
     public List<TableTestProgram> programs() {
         return List.of(

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/SemanticTestBase.java

@@ -21,6 +21,7 @@
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableConfig;
 import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
 import org.apache.flink.table.planner.factories.TestValuesTableFactory;
 import org.apache.flink.table.test.program.ConfigOptionTestStep;
 import org.apache.flink.table.test.program.FailingSqlTestStep;
@@ -157,7 +158,11 @@ protected void runStep(TestStep testStep, TableEnvironment env) throws Exception
      *
      * <p>Default implementation is a no-op. Subclasses can override to set specific options.
      */
-    protected void applyDefaultEnvironmentOptions(TableConfig config) {}
+    protected void applyDefaultEnvironmentOptions(TableConfig config) {
+        config.set(
+                OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
+                OptimizerConfigOptions.NonDeterministicUpdateStrategy.TRY_RESOLVE);
+    }
 
     private static Map<String, String> createSourceOptions(
             SourceTestStep sourceTestStep, String id) {