[ML] Convert encodeIds flag in TreeEncoder to configuration property

wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MLContext.java

@@ -63,8 +63,8 @@ public void execute(final WayangPlan wayangPlan, final String... udfJars) {
             final Optional<String> originalOption = config.getOptionalStringProperty("wayang.ml.experience.original");
 
             final OneHotMappings mappings = new OneHotMappings();
-            final TreeEncoder encoder = new TreeEncoder(mappings);
-            final String original = originalOption.orElse(encoder.encode(wayangPlan, wayangJob.getOptimizationContext(), false).toString());
+            final TreeEncoder encoder = new TreeEncoder(mappings, config);
+            final String original = originalOption.orElse(encoder.encode(wayangPlan, wayangJob.getOptimizationContext()).toString());
 
             final Optional<String> choicesOption = config
                     .getOptionalStringProperty("wayang.ml.experience.with-platforms");

wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/MachineLearning.java

@@ -18,6 +18,7 @@
 
 package org.apache.wayang.ml;
 
+import java.io.InputStream;
 import java.util.Collection;
 import java.util.Collections;
 
@@ -26,6 +27,7 @@
 import org.apache.wayang.core.optimizer.channels.ChannelConversion;
 import org.apache.wayang.core.platform.Platform;
 import org.apache.wayang.core.plugin.Plugin;
+import org.apache.wayang.core.util.ReflectionUtils;
 import org.apache.wayang.java.platform.JavaPlatform;
 import org.apache.wayang.spark.platform.SparkPlatform;
 
@@ -34,6 +36,8 @@
  */
 public class MachineLearning {
 
+    private static final String DEFAULT_CONFIG_FILE = "wayang-ml-defaults.properties";
+
     /**
      * Enables use with the {@link JavaPlatform} and {@link SparkPlatform}.
      */
@@ -56,9 +60,47 @@ public Collection<ChannelConversion> getChannelConversions() {
 
         @Override
         public void setProperties(final Configuration configuration) {
+            try (final InputStream is = ReflectionUtils.loadResource(DEFAULT_CONFIG_FILE)) {
+                if (is != null) {
+                    final java.util.Properties properties = new java.util.Properties();
+                    properties.load(is);
+                    for (final java.util.Map.Entry<Object, Object> propertyEntry : properties.entrySet()) {
+                        final String key = propertyEntry.getKey().toString();
+                        final String value = propertyEntry.getValue().toString();
+                        if (isConfigured(configuration, key)) {
+                            continue;
+                        }
+                        configuration.setProperty(key, value);
+                    }
+                }
+            } catch (java.io.IOException e) {
+                throw new org.apache.wayang.core.api.exception.WayangException("Could not load ML default configuration.", e);
+            }
         }
     };
 
+    private static boolean isConfigured(final Configuration configuration, final String key) {
+        return isExplicitlyConfigured(configuration, key);
+    }
+
+    private static boolean isExplicitlyConfigured(final Configuration configuration, final String key) {
+        if (configuration == null) {
+            return false;
+        }
+        if (configuration == Configuration.getDefaultConfiguration()) {
+            return configuration.getProperties() != null
+                    && configuration.getProperties().provideLocally(key) != null;
+        }
+        Configuration current = configuration;
+        while (current != null && current != Configuration.getDefaultConfiguration()) {
+            if (current.getProperties() != null && current.getProperties().provideLocally(key) != null) {
+                return true;
+            }
+            current = current.getParent();
+        }
+        return false;
+    }
+
     /**
      * Retrieve a {@link Plugin} to use xxx on the
      *

wayang-plugins/wayang-ml/src/main/java/org/apache/wayang/ml/encoding/TreeEncoder.java

@@ -27,6 +27,7 @@
 import java.util.Map;
 import java.util.Queue;
 
+import org.apache.wayang.core.api.Configuration;
 import org.apache.wayang.core.api.exception.WayangException;
 import org.apache.wayang.core.optimizer.OptimizationContext;
 import org.apache.wayang.core.optimizer.enumeration.PlanImplementation;
@@ -40,27 +41,100 @@
 import org.apache.wayang.core.platform.Junction;
 
 public class TreeEncoder {
+
+    public static final String ENCODE_IDS_PROPERTY = "wayang.ml.encoding.encode-ids";
+
+    public static final String ENCODE_IDS_SHORTHAND_PROPERTY = "wayang.ml.encode-ids";
+
     private final OneHotMappings mappings;
 
+    private Configuration configuration;
+
     public TreeEncoder(final OneHotMappings mappings) {
+        this(mappings, null);
+    }
+
+    public TreeEncoder(final OneHotMappings mappings, final Configuration configuration) {
         this.mappings = mappings;
+        this.configuration = configuration;
     }
 
     public OneHotMappings getMappings() {
         return this.mappings;
     }
 
+    public Configuration getConfiguration() {
+        return this.configuration;
+    }
+
+    public void setConfiguration(final Configuration configuration) {
+        this.configuration = configuration;
+    }
+
+    /**
+     * Resolves whether operator IDs should be encoded. The configuration hierarchy is walked from the given
+     * (most specific, e.g. job-level) configuration up to its ancestors; the closest level that sets either
+     * {@link #ENCODE_IDS_PROPERTY} or {@link #ENCODE_IDS_SHORTHAND_PROPERTY} wins, so a child can override an
+     * inherited value even if it uses the other spelling. If both are set on the very same level, a spelling
+     * that merely mirrors the inherited value (e.g., cached by an earlier read through the child) loses
+     * against the other one; otherwise the canonical property takes precedence. Defaults to {@code false}.
+     */
+    public static boolean resolveEncodeIds(final Configuration config) {
+        for (Configuration level = config; level != null; level = level.getParent()) {
+            final String canonical = level.getProperties().provideLocally(ENCODE_IDS_PROPERTY);
+            final String shorthand = level.getProperties().provideLocally(ENCODE_IDS_SHORTHAND_PROPERTY);
+            if (canonical == null && shorthand == null) {
+                continue;
+            }
+            if (canonical == null) {
+                return Boolean.parseBoolean(shorthand);
+            }
+            if (shorthand == null) {
+                return Boolean.parseBoolean(canonical);
+            }
+            final Configuration parent = level.getParent();
+            final boolean canonicalMirrorsParent = canonical.equals(inheritedValue(parent, ENCODE_IDS_PROPERTY));
+            final boolean shorthandMirrorsParent = shorthand.equals(inheritedValue(parent, ENCODE_IDS_SHORTHAND_PROPERTY));
+            return canonicalMirrorsParent && !shorthandMirrorsParent
+                    ? Boolean.parseBoolean(shorthand)
+                    : Boolean.parseBoolean(canonical);
+        }
+        return false;
+    }
+
+    /**
+     * Looks up the value that {@code level} or its closest ancestor defines for {@code key} without touching
+     * any property caches.
+     */
+    private static String inheritedValue(final Configuration level, final String key) {
+        for (Configuration current = level; current != null; current = current.getParent()) {
+            final String value = current.getProperties().provideLocally(key);
+            if (value != null) {
+                return value;
+            }
+        }
+        return null;
+    }
+
+    private boolean resolveEncodeIds(final OptimizationContext optimizationContext) {
+        final Configuration config = this.configuration != null
+                ? this.configuration
+                : (optimizationContext != null ? optimizationContext.getConfiguration() : null);
+        return resolveEncodeIds(config);
+    }
+
     public TreeNode encode(final PlanImplementation plan) {
+        return this.encode(plan, this.resolveEncodeIds(plan.getOptimizationContext()));
+    }
+
+    public TreeNode encode(final PlanImplementation plan, final boolean encodeIds) {
         final List<TreeNode> result = new ArrayList<TreeNode>();
 
         final HashMap<Operator, Collection<Operator>> tree = new HashMap<>();
         final List<ExecutionOperator> sinks = plan.getOperators().stream().filter(Operator::isSink).toList();
 
         final Map<OutputSlot<?>, Junction> junctions = plan.getJunctions();
 
-        // TODO: convert to config
-        final boolean encodeIds = false;
-
         for (final Operator sink : sinks) {
             final TreeNode sinkNode = traversePIOperator(sink, plan.getOptimizationContext(), encodeIds, junctions,
                     tree);
@@ -77,6 +151,10 @@ public TreeNode encode(final PlanImplementation plan) {
         return resultNode;
     }
 
+    public TreeNode encode(final WayangPlan plan, final OptimizationContext optimizationContext) {
+        return this.encode(plan, optimizationContext, this.resolveEncodeIds(optimizationContext));
+    }
+
     public TreeNode encode(final WayangPlan plan, final OptimizationContext optimizationContext,
             final boolean encodeIds) {
         final List<TreeNode> result = new ArrayList<TreeNode>();

wayang-plugins/wayang-ml/src/main/resources/wayang-ml-defaults.properties

@@ -73,4 +73,7 @@ wayang.ml.experience.enabled = false
 wayang.ml.executions.file = /var/www/html/data/executions.txt
 wayang.ml.optimizations.file = /var/www/html/data/optmizations.txt
 wayang.ml.experience.file = /var/www/html/data/experience/experience-vae.txt
+# Not set on purpose (TreeEncoder defaults to false); a default here would shadow inherited settings.
+# Use wayang.ml.encoding.encode-ids or the shorthand wayang.ml.encode-ids.
+# wayang.ml.encoding.encode-ids = false
 org.apache.logging.log4j.level = INFO