Fixed errors

Author: alex268

src/main/java/tech/ydb/app/Application.java

@@ -74,7 +74,7 @@ public void run(String... args) {
         for (CdcReader reader: readers) {
             sessionPoolSize += reader.getWriter().getThreadsCount();
         }
-        ydb.updatePoolSize(Math.min(sessionPoolSize, 50));
+        ydb.updatePoolSize(Math.max(sessionPoolSize, 50));
 
         for (CdcReader reader: readers) {
             reader.start();

src/main/java/tech/ydb/app/CdcMsgParser.java

@@ -158,8 +158,8 @@ private Result<Supplier<YqlQuery>> findDeleteQuery(TableDescription source) {
 
         @SuppressWarnings("null")
         private Result<Supplier<YqlQuery>> validate(TableDescription source, XmlConfig.Query query, boolean keysOnly) {
-            String queryText = query.getText().trim();
-            Result<DataQuery> parsed = ydb.parseQuery(queryText);
+            String text = query.getText().trim();
+            Result<DataQuery> parsed = ydb.parseQuery(text);
             if (!parsed.isSuccess()) {
                 logger.error("Can't parse query for consumer {}, got status {}", cdc.getConsumer(), parsed.getStatus());
                 return parsed.map(null);
@@ -222,24 +222,33 @@ private Result<Supplier<YqlQuery>> validate(TableDescription source, XmlConfig.Q
             }
 
             List<String> keys = source.getPrimaryKeys();
-            if (query.getUpsertTo() != null && !query.getUpsertTo().trim().isEmpty()) {
-                String execute = "UPSERT INTO `" + query.getUpsertTo().trim() + "` ";
-                return Result.success(YqlQuery.readAndExecuteYql(queryText, execute, keys, paramName, structType, cdc));
-            }
-            if (query.getDeleteFrom() != null && !query.getDeleteFrom().trim().isEmpty()) {
-                String execute = "DELETE FROM `" + query.getDeleteFrom().trim() + "` ON ";
-                return Result.success(YqlQuery.readAndExecuteYql(queryText, execute, keys, paramName, structType, cdc));
-            }
-            if (query.getUpdateOn() != null && !query.getUpdateOn().trim().isEmpty()) {
-                String execute = "UPDATE `" + query.getUpdateOn().trim() + "` ON ";
-                return Result.success(YqlQuery.readAndExecuteYql(queryText, execute, keys, paramName, structType, cdc));
-            }
-            if (query.getInsertTo() != null && !query.getInsertTo().trim().isEmpty()) {
-                String execute = "INSERT INTO `" + query.getInsertTo().trim() + "` ";
-                return Result.success(YqlQuery.readAndExecuteYql(queryText, execute, keys, paramName, structType, cdc));
+            if (query.getActionTable() != null && !query.getActionTable().trim().isEmpty()) {
+                String actionTable = query.getActionTable().trim();
+                String action = query.getActionMode();
+                if ("upsertInto".equalsIgnoreCase(action)) {
+                    String execute = "UPSERT INTO `" + actionTable + "` ";
+                    return Result.success(YqlQuery.readAndExecuteYql(text, execute, keys, paramName, structType, cdc));
+                }
+                if ("deleteFrom".equalsIgnoreCase(action)) {
+                    String execute = "DELETE FROM `" + actionTable + "` ON ";
+                    return Result.success(YqlQuery.readAndExecuteYql(text, execute, keys, paramName, structType, cdc));
+                }
+                if ("updateOn".equalsIgnoreCase(action)) {
+                    String execute = "UPDATE `" + actionTable + "` ON ";
+                    return Result.success(YqlQuery.readAndExecuteYql(text, execute, keys, paramName, structType, cdc));
+                }
+                if ("insertInto".equalsIgnoreCase(action)) {
+                    String execute = "INSERT INTO `" + actionTable + "` ";
+                    return Result.success(YqlQuery.readAndExecuteYql(text, execute, keys, paramName, structType, cdc));
+                }
+
+                return Result.fail(Status.of(StatusCode.CLIENT_INTERNAL_ERROR, Issue.of(
+                        "Uknown actionName " + action + ", expected upsertInto/deleteFrom/updateOn/insertInto",
+                        Issue.Severity.ERROR
+                )));
             }
 
-            return Result.success(YqlQuery.executeYql(queryText, keys, paramName, structType, cdc));
+            return Result.success(YqlQuery.executeYql(text, keys, paramName, structType, cdc));
         }
     }
 

src/main/java/tech/ydb/app/XmlConfig.java

@@ -36,17 +36,11 @@ public static class Query {
         @XmlAttribute(name = "id", required = true)
         private String id;
 
-        @XmlAttribute(name = "upsertTo")
-        private String upsertTo;
+        @XmlAttribute(name = "actionMode")
+        private String actionMode;
 
-        @XmlAttribute(name = "insertTo")
-        private String insertTo;
-
-        @XmlAttribute(name = "updateOn")
-        private String updateOn;
-
-        @XmlAttribute(name = "deleteFrom")
-        private String deleteFrom;
+        @XmlAttribute(name = "actionTable")
+        private String actionTable;
 
 //        @XmlAttribute(name = "batchSize")
 //        private Integer batchSize;
@@ -70,21 +64,14 @@ public String getText() {
             return this.text;
         }
 
-        public String getUpsertTo() {
-            return this.upsertTo;
+        public String getActionMode() {
+            return this.actionMode;
         }
 
-        public String getInsertTo() {
-            return this.insertTo;
+        public String getActionTable() {
+            return this.actionTable;
         }
 
-        public String getDeleteFrom() {
-            return this.deleteFrom;
-        }
-
-        public String getUpdateOn() {
-            return this.updateOn;
-        }
 //        public int getBatchSize() {
 //            if (batchSize == null) {
 //                return DEFAULT_BATCH_SIZE;

src/main/java/tech/ydb/app/YqlQuery.java

@@ -216,8 +216,8 @@ public Status execute(YdbService ydb) {
 
                 StructType type = resultSetToType(rs);
                 Value<?> values = ListType.of(type).newValue(resultSetToValues(rs, type));
-                Params executePrms = Params.of("$input", values);
-                String executeQuery = "DECLARE $input AS " + type + "; " + query + " SELECT FROM AS_TABLE($input);";
+                Params executePrms = Params.of("$b", values);
+                String executeQuery = "DECLARE $b AS List<" + type + ">; " + query + " SELECT * FROM AS_TABLE($b);";
                 return ydb.executeYqlQuery(executeQuery, executePrms, timeout);
             }
         };
@@ -239,7 +239,7 @@ private static List<Value<?>> resultSetToValues(ResultSetReader rs, StructType t
         while (rs.next()) {
             Value<?>[] row = new Value[type.getMembersCount()];
             for (int idx = 0; idx < type.getMembersCount(); idx += 1) {
-                row[idx] = rs.getColumn(type.getMemberIndex(type.getMemberName(idx))).getValue();
+                row[idx] = rs.getColumn(type.getMemberName(idx)).getValue();
             }
             values.add(type.newValueUnsafe(row));
         }