SQL: Register missing processors (#35121)

Add registration (and tests) for missing processors in the serialization
chain.

Fix #35119

(cherry picked from commit f87a534)

Author: costin

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/function/scalar/Processors.java

@@ -25,9 +25,14 @@
 import org.elasticsearch.xpack.sql.expression.gen.processor.ConstantProcessor;
 import org.elasticsearch.xpack.sql.expression.gen.processor.HitExtractorProcessor;
 import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+import org.elasticsearch.xpack.sql.expression.predicate.IsNotNullProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.logical.BinaryLogicProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.logical.NotProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.BinaryArithmeticProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.arithmetic.UnaryArithmeticProcessor;
 import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.BinaryComparisonProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.operator.comparison.InProcessor;
+import org.elasticsearch.xpack.sql.expression.predicate.regex.RegexProcessor;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -49,13 +54,23 @@ public static List<NamedWriteableRegistry.Entry> getNamedWriteables() {
         entries.add(new Entry(Processor.class, CastProcessor.NAME, CastProcessor::new));
         entries.add(new Entry(Processor.class, ChainingProcessor.NAME, ChainingProcessor::new));
 
-        // comparators
-        entries.add(new Entry(Processor.class, BinaryComparisonProcessor.NAME, BinaryComparisonProcessor::new));
+        // logical
+        entries.add(new Entry(Processor.class, BinaryLogicProcessor.NAME, BinaryLogicProcessor::new));
+        entries.add(new Entry(Processor.class, NotProcessor.NAME, NotProcessor::new));
+        // null
+        entries.add(new Entry(Processor.class, IsNotNullProcessor.NAME, IsNotNullProcessor::new));
 
         // arithmetic
         entries.add(new Entry(Processor.class, BinaryArithmeticProcessor.NAME, BinaryArithmeticProcessor::new));
         entries.add(new Entry(Processor.class, UnaryArithmeticProcessor.NAME, UnaryArithmeticProcessor::new));
         entries.add(new Entry(Processor.class, BinaryMathProcessor.NAME, BinaryMathProcessor::new));
+        // comparators
+        entries.add(new Entry(Processor.class, BinaryComparisonProcessor.NAME, BinaryComparisonProcessor::new));
+        entries.add(new Entry(Processor.class, InProcessor.NAME, InProcessor::new));
+        // regex
+        entries.add(new Entry(Processor.class, RegexProcessor.NAME, RegexProcessor::new));
+
+
         // datetime
         entries.add(new Entry(Processor.class, DateTimeProcessor.NAME, DateTimeProcessor::new));
         entries.add(new Entry(Processor.class, NamedDateTimeProcessor.NAME, NamedDateTimeProcessor::new));

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/IsNotNullProcessor.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.sql.expression.predicate;
 
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
 
@@ -18,6 +19,8 @@ public class IsNotNullProcessor implements Processor {
 
     private IsNotNullProcessor() {}
 
+    public IsNotNullProcessor(StreamInput in) throws IOException {}
+
     @Override
     public String getWriteableName() {
         return NAME;

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/expression/predicate/logical/NotProcessor.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.sql.expression.predicate.logical;
 
+import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.sql.SqlIllegalArgumentException;
 import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
@@ -19,6 +20,8 @@ public class NotProcessor implements Processor {
 
     private NotProcessor() {}
 
+    public NotProcessor(StreamInput in) throws IOException {}
+
     @Override
     public String getWriteableName() {
         return NAME;

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/expression/ProcessorTests.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+package org.elasticsearch.xpack.sql.expression;
+
+import org.elasticsearch.common.io.stream.NamedWriteable;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.sql.expression.function.scalar.Processors;
+import org.elasticsearch.xpack.sql.expression.gen.processor.Processor;
+import org.elasticsearch.xpack.sql.tree.NodeSubclassTests;
+import org.junit.BeforeClass;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+
+import static java.util.stream.Collectors.toCollection;
+
+
+public class ProcessorTests extends ESTestCase {
+
+    private static List<Class<? extends Processor>> processors;
+
+    @BeforeClass
+    public static void init() throws Exception {
+        processors = NodeSubclassTests.subclassesOf(Processor.class);
+    }
+
+
+    public void testProcessorRegistration() throws Exception {
+        LinkedHashSet<String> registered = Processors.getNamedWriteables().stream()
+                .map(e -> e.name)
+                .collect(toCollection(LinkedHashSet::new));
+
+        // discover available processors
+        int missing = processors.size() - registered.size();
+
+
+        if (missing > 0) {
+            List<String> notRegistered = new ArrayList<>();
+            for (Class<? extends Processor> proc : processors) {
+                String procName = proc.getName();
+                assertTrue(procName + " does NOT implement NamedWriteable", NamedWriteable.class.isAssignableFrom(proc));
+                Field name = null;
+                String value = null;
+                try {
+                    name = proc.getField("NAME");
+                } catch (Exception ex) {
+                    fail(procName + " does NOT provide a NAME field\n" + ex);
+                }
+                try {
+                    value = name.get(proc).toString();
+                } catch (Exception ex) {
+                    fail(procName + " does NOT provide a static NAME field\n" + ex);
+                }
+                if (!registered.contains(value)) {
+                    notRegistered.add(procName);
+                }
+            }
+            
+            fail(missing + " processor(s) not registered : " + notRegistered);
+        } else {
+            assertEquals("Detection failed: discovered more registered processors than classes", 0, missing);
+        }
+    }
+}

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/tree/NodeSubclassTests.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.sql.tree;
 
 import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.PathUtils;
 import org.elasticsearch.test.ESTestCase;
@@ -585,7 +586,7 @@ public static <T extends Node<?>> T makeNode(Class<? extends T> nodeClass) throw
     /**
      * Find all subclasses of a particular class.
      */
-    private static <T> List<Class<? extends T>> subclassesOf(Class<T> clazz) throws IOException {
+    public static <T> List<Class<? extends T>> subclassesOf(Class<T> clazz) throws IOException {
         @SuppressWarnings("unchecked") // The map is built this way
         List<Class<? extends T>> lookup = (List<Class<? extends T>>) subclassCache.get(clazz);
         if (lookup != null) {