Merge branch 'master' of https://github.com/apache/iotdb into pipe-api

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/ainode/it/AINodeConcurrentForecastIT.java

@@ -36,8 +36,9 @@
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.Arrays;
+import java.util.List;
 
-import static org.apache.iotdb.ainode.utils.AINodeTestUtils.BUILTIN_LTSM_MAP;
 import static org.apache.iotdb.ainode.utils.AINodeTestUtils.checkModelNotOnSpecifiedDevice;
 import static org.apache.iotdb.ainode.utils.AINodeTestUtils.checkModelOnSpecifiedDevice;
 import static org.apache.iotdb.ainode.utils.AINodeTestUtils.concurrentInference;
@@ -48,6 +49,11 @@ public class AINodeConcurrentForecastIT {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AINodeConcurrentForecastIT.class);
 
+  private static final List<AINodeTestUtils.FakeModelInfo> MODEL_LIST =
+      Arrays.asList(
+          new AINodeTestUtils.FakeModelInfo("sundial", "sundial", "builtin", "active"),
+          new AINodeTestUtils.FakeModelInfo("timer_xl", "timer", "builtin", "active"));
+
   private static final String FORECAST_TABLE_FUNCTION_SQL_TEMPLATE =
       "SELECT * FROM FORECAST(model_id=>'%s', targets=>(SELECT time,s FROM root.AI) ORDER BY time, output_length=>%d)";
 
@@ -78,7 +84,7 @@ private static void prepareDataForTableModel() throws SQLException {
 
   @Test
   public void concurrentGPUForecastTest() throws SQLException, InterruptedException {
-    for (AINodeTestUtils.FakeModelInfo modelInfo : BUILTIN_LTSM_MAP.values()) {
+    for (AINodeTestUtils.FakeModelInfo modelInfo : MODEL_LIST) {
       concurrentGPUForecastTest(modelInfo);
     }
   }

iotdb-api/pipe-api/pom.xml

@@ -35,6 +35,11 @@
             <version>${tsfile.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <profiles>
         <profile>

iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParameters.java

@@ -367,24 +367,35 @@ public PipeParameters addOrReplaceEquivalentAttributesWithClone(final PipeParame
 
   private static class KeyReducer {
 
-    private static final Set<String> PREFIXES = new HashSet<>();
+    private static final Set<String> FIRST_PREFIXES = new HashSet<>();
+    private static final Set<String> SECOND_PREFIXES = new HashSet<>();
 
     static {
-      PREFIXES.add("extractor.");
-      PREFIXES.add("source.");
-      PREFIXES.add("processor.");
-      PREFIXES.add("connector.");
-      PREFIXES.add("sink.");
+      FIRST_PREFIXES.add("extractor.");
+      FIRST_PREFIXES.add("source.");
+      FIRST_PREFIXES.add("processor.");
+      FIRST_PREFIXES.add("connector.");
+      FIRST_PREFIXES.add("sink.");
+
+      SECOND_PREFIXES.add("opcua.");
     }
 
-    static String reduce(final String key) {
+    static String reduce(String key) {
       if (key == null) {
         return null;
       }
-      final String lowerCaseKey = key.toLowerCase();
-      for (final String prefix : PREFIXES) {
+      String lowerCaseKey = key.toLowerCase();
+      for (final String prefix : FIRST_PREFIXES) {
+        if (lowerCaseKey.startsWith(prefix)) {
+          key = key.substring(prefix.length());
+          lowerCaseKey = lowerCaseKey.substring(prefix.length());
+          break;
+        }
+      }
+      for (final String prefix : SECOND_PREFIXES) {
         if (lowerCaseKey.startsWith(prefix)) {
-          return key.substring(prefix.length());
+          key = key.substring(prefix.length());
+          break;
         }
       }
       return key;

iotdb-api/pipe-api/src/test/java/org/apache/iotdb/pipe/api/customizer/parameter/PipeParametersTest.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.pipe.api.customizer.parameter;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+
+public class PipeParametersTest {
+
+  @Test
+  public void keyReducerTest() {
+    final PipeParameters parameters = new PipeParameters(new HashMap<>());
+    parameters.addAttribute("sink.opcua.with-quality", "false");
+
+    Assert.assertEquals(false, parameters.getBoolean("with-quality"));
+    Assert.assertEquals(false, parameters.getBoolean("opcua.with-quality"));
+
+    // Invalid
+    parameters.addAttribute("sink.source.opcua.value-name", "false");
+    parameters.addAttribute("opcua.sink.value-name", "false");
+    Assert.assertNull(parameters.getString("value-name"));
+  }
+}

iotdb-core/ainode/iotdb/ainode/core/model/model_info.py

@@ -116,6 +116,7 @@ def __repr__(self):
             "AutoConfig": "configuration_timer.TimerConfig",
             "AutoModelForCausalLM": "modeling_timer.TimerForPrediction",
         },
+        _transformers_registered=True,
     ),
     "sundial": ModelInfo(
         model_id="sundial",
@@ -128,5 +129,6 @@ def __repr__(self):
             "AutoConfig": "configuration_sundial.SundialConfig",
             "AutoModelForCausalLM": "modeling_sundial.SundialForPrediction",
         },
+        _transformers_registered=True,
     ),
 }

iotdb-core/ainode/iotdb/ainode/core/model/model_storage.py

@@ -196,9 +196,12 @@ def _callback_model_download_result(self, future, model_id: str):
                     if os.path.exists(config_path):
                         with open(config_path, "r", encoding="utf-8") as f:
                             config = json.load(f)
-                        if model_info.model_type == "":
-                            model_info.model_type = config.get("model_type", "")
-                        model_info.auto_map = config.get("auto_map", None)
+                        model_info.model_type = config.get(
+                            "model_type", model_info.model_type
+                        )
+                        model_info.auto_map = config.get(
+                            "auto_map", model_info.auto_map
+                        )
                     logger.info(
                         f"Model {model_id} downloaded successfully and is ready to use."
                     )