clazz) {
+ try {
+ return clazz.cast(actual.get(nth).value);
+ } catch (ClassCastException e) {
+ throw new TypeException(null, clazz, actual.get(nth).type, actual.get(nth).value);
+ }
+ }
+
+ public interface Arity {
+ void check(String methodName, List params);
+
+ Arity NULLARY = Arity.of(0);
+ Arity UNARY = Arity.of(1);
+ Arity BINARY = Arity.of(2);
+
+ static Arity of(int count) {
+ return (methodName, params) -> {
+ if (params.size() != count) {
+ throw new ArityException(null, methodName, count, params.size());
+ }
+ };
+ }
+
+ static Arity min(int count) {
+ return (methodName, params) -> {
+ if (params.size() < count) {
+ throw new ArityException(null, "wrong number of arguments in call to '" + methodName
+ + "'. Expected at least " + count + " got " + params.size() + ".");
+ }
+ };
+ }
+
+ static Arity max(int count) {
+ return (methodName, params) -> {
+ if (params.size() > count) {
+ throw new ArityException(null, "wrong number of arguments in call to '" + methodName
+ + "'. Expected at most " + count + " got " + params.size() + ".");
+ }
+ };
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/Table.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/Table.java
new file mode 100644
index 00000000000000..57aad8e1d1cf5b
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/Table.java
@@ -0,0 +1,224 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/objects/Table.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.objects;
+
+import org.apache.doris.hplsql.ColumnDefinition;
+import org.apache.doris.hplsql.Row;
+import org.apache.doris.hplsql.Var;
+import org.apache.doris.hplsql.executor.QueryResult;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Oracle's PL/SQL Table/associative array.
+ *
+ * Tables can be modelled after a corresponding Hive table or they can be created manually.
+ *
+ * 1. Model the table after the emp Hive table
+ * TYPE t_tab IS TABLE OF emp%ROWTYPE INDEX BY BINARY_INTEGER;
+ *
+ * 2. Model the table after a column of a Hive table (emp.name). This table will hold a single column only.
+ * TYPE t_tab IS TABLE OF emp.name%TYPE INDEX BY BINARY_INTEGER;
+ *
+ * 3. Or you can specify the column manually. This table will hold one column only.
+ * TYPE t_tab IS TABLE OF NUMBER INDEX BY BINARY_INTEGER;
+ *
+ * In the first case the values will be records where each key in the record matches the columns to the corresponding
+ * table.
+ * tab(key).col_name;
+ *
+ * In the last two cases the values will represent scalars, but they stored in a record with a single key.
+ * tab(key)
+ *
+ * Iteration logic uses the first/last next and prior methods.
+ * First/last returns a key, next/prior gives back the next or previous key based on the key passed in.
+ */
+public class Table implements HplObject {
+ private final TableClass hplClass;
+ private final Map rows = new HashMap<>();
+ private Object lastKey = null;
+ private Object firstKey = null;
+
+ public Table(TableClass hplClass) {
+ this.hplClass = hplClass;
+ }
+
+ public void populate(QueryResult query, long rowIndex, int columnIndex) {
+ if (hplClass().rowType()) {
+ putRow(rowIndex, query);
+ } else {
+ putColumn(rowIndex, query, columnIndex);
+ }
+ }
+
+ public void putRow(Object key, QueryResult result) {
+ put(key, readRow(result));
+ }
+
+ public void putColumn(Object key, QueryResult query, int columnIndex) {
+ put(key, readColumn(query, columnIndex));
+ }
+
+ public void put(Object key, Row row) {
+ Value existing = rows.get(key);
+ if (existing != null) {
+ existing.row = row;
+ } else {
+ if (lastKey != null) {
+ rows.get(lastKey).nextKey = key;
+ }
+ rows.put(key, new Value(row, lastKey));
+ lastKey = key;
+ if (firstKey == null) {
+ firstKey = key;
+ }
+ }
+ }
+
+ private Row readRow(QueryResult result) {
+ Row row = new Row();
+ int idx = 0;
+ for (ColumnDefinition column : hplClass.columns()) {
+ Var var = new Var(column.columnName(), column.columnType().typeString(), (Integer) null, null, null);
+ var.setValue(result, idx);
+ row.addColumn(column.columnName(), column.columnTypeString(), var);
+ idx++;
+ }
+ return row;
+ }
+
+ private Row readColumn(QueryResult result, int columnIndex) {
+ Row row = new Row();
+ ColumnDefinition column = hplClass.columns().get(0);
+ Var var = new Var(column.columnName(), column.columnType().typeString(), (Integer) null, null, null);
+ var.setValue(result, columnIndex);
+ row.addColumn(column.columnName(), column.columnTypeString(), var);
+ return row;
+ }
+
+ public Row at(Object key) {
+ Value value = rows.get(key);
+ return value == null ? null : value.row;
+ }
+
+ public boolean removeAt(Object key) {
+ Value value = rows.remove(key);
+ if (value != null) {
+ updateLinks(key, value.nextKey, value.prevKey);
+ }
+ return value != null;
+ }
+
+ private void updateLinks(Object deletedKey, Object nextKey, Object prevKey) {
+ if (prevKey != null) {
+ rows.get(prevKey).nextKey = nextKey;
+ }
+ if (nextKey != null) {
+ rows.get(nextKey).prevKey = prevKey;
+ }
+ if (deletedKey.equals(firstKey)) {
+ firstKey = nextKey;
+ }
+ if (deletedKey.equals(lastKey)) {
+ lastKey = prevKey;
+ }
+ }
+
+ public void removeFromTo(Object fromKey, Object toKey) {
+ Object current = fromKey;
+ while (current != null && !current.equals(toKey)) {
+ Object next = nextKey(current);
+ removeAt(current);
+ current = next;
+ }
+ if (current != null) {
+ removeAt(current);
+ }
+ }
+
+ public void removeAll() {
+ lastKey = null;
+ firstKey = null;
+ rows.clear();
+ }
+
+ public Object nextKey(Object key) {
+ Value value = rows.get(key);
+ return value == null ? null : value.nextKey;
+ }
+
+ public Object priorKey(Object key) {
+ Value value = rows.get(key);
+ return value == null ? null : value.prevKey;
+ }
+
+ public Object firstKey() {
+ return firstKey;
+ }
+
+ public Object lastKey() {
+ return lastKey;
+ }
+
+ public boolean existsAt(Object key) {
+ return rows.get(key) != null;
+ }
+
+ public int count() {
+ return rows.size();
+ }
+
+ @Override
+ public TableClass hplClass() {
+ return hplClass;
+ }
+
+ private static class Value {
+ private Row row;
+ private Object prevKey;
+ private Object nextKey;
+
+ public Value(Row row, Object prevKey) {
+ this.row = row;
+ this.prevKey = prevKey;
+ }
+
+ public void setPrevKey(Object prevKey) {
+ this.prevKey = prevKey;
+ }
+
+ public void setNextKey(Object nextKey) {
+ this.nextKey = nextKey;
+ }
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("Table{");
+ sb.append("hplClass=").append(hplClass.getClass());
+ sb.append(", size=").append(count());
+ sb.append(", lastKey=").append(lastKey);
+ sb.append(", firstKey=").append(firstKey);
+ sb.append('}');
+ return sb.toString();
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/TableClass.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/TableClass.java
new file mode 100644
index 00000000000000..ef75f6007516d4
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/TableClass.java
@@ -0,0 +1,135 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/objects/TableClass.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.objects;
+
+import org.apache.doris.hplsql.ColumnDefinition;
+import org.apache.doris.hplsql.Row;
+import org.apache.doris.hplsql.Var;
+import org.apache.doris.hplsql.objects.MethodParams.Arity;
+
+import java.util.List;
+
+
+public class TableClass implements HplClass {
+ private static final MethodDictionary methodDictionary = new MethodDictionary<>();
+ private final String typeName;
+ private final List columns;
+ private final boolean rowType;
+
+ static {
+ methodDictionary.put("first", (self, args) -> {
+ Arity.NULLARY.check("first", args);
+ return wrap(self.firstKey());
+ });
+ methodDictionary.put("last", (self, args) -> {
+ Arity.NULLARY.check("last", args);
+ return wrap(self.lastKey());
+ });
+ methodDictionary.put("next", (self, args) -> {
+ Long key = new MethodParams("next", args, Arity.UNARY).longAt(0);
+ return wrap(self.nextKey(key));
+ });
+ methodDictionary.put("prior", (self, args) -> {
+ Long key = new MethodParams("prior", args, Arity.UNARY).longAt(0);
+ return wrap(self.priorKey(key));
+ });
+ methodDictionary.put("count", (self, args) -> {
+ Arity.NULLARY.check("count", args);
+ return new Var(Long.valueOf(self.count()));
+ });
+ methodDictionary.put("exists", (self, args) -> {
+ Long key = new MethodParams("exists", args, Arity.UNARY).longAt(0);
+ return new Var(self.existsAt(key));
+ });
+ methodDictionary.put("delete", (self, args) -> {
+ Arity.max(3).check("delete", args);
+ if (args.isEmpty()) {
+ self.removeAll();
+ } else if (args.size() == 1) {
+ self.removeAt(args.get(0).value);
+ } else {
+ self.removeFromTo(args.get(0).value, args.get(1).value);
+ }
+ return null;
+ });
+ methodDictionary.put(MethodDictionary.__GETITEM__, (self, args) -> {
+ Long key = new MethodParams(MethodDictionary.__GETITEM__, args, Arity.UNARY).longAt(0);
+ Row row = self.at(key);
+ if (row == null) {
+ return Var.Null;
+ }
+ if (self.hplClass().rowType()) {
+ Var var = new Var();
+ var.setType(Var.Type.ROW.name());
+ var.setValue(row);
+ return var;
+ }
+ return row.getValue(0);
+ });
+ methodDictionary.put(MethodDictionary.__SETITEM__, (self, args) -> {
+ MethodParams params = new MethodParams(MethodDictionary.__SETITEM__, args, Arity.BINARY);
+ long key = params.longAt(0);
+ if (self.hplClass().rowType()) {
+ self.put(key, params.rowAt(1));
+ } else { // single column
+ Row row = new Row();
+ row.addColumn(
+ self.hplClass().columns().get(0).columnName(),
+ self.hplClass().columns.get(0).columnTypeString(),
+ args.get(1));
+ self.put(key, row);
+ }
+ return Var.Null;
+ });
+ }
+
+ private static Var wrap(Object result) {
+ return result != null ? new Var((Long) result) : Var.Null;
+ }
+
+ public TableClass(String typeName, List columns, boolean rowType) {
+ this.typeName = typeName;
+ this.columns = columns;
+ this.rowType = rowType;
+ }
+
+ public String typeName() {
+ return typeName;
+ }
+
+ public List columns() {
+ return columns;
+ }
+
+ @Override
+ public Table newInstance() {
+ return new Table(this);
+ }
+
+ @Override
+ public MethodDictionary methodDictionary() {
+ return methodDictionary;
+ }
+
+ public boolean rowType() {
+ return rowType;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/UtlFile.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/UtlFile.java
new file mode 100644
index 00000000000000..9d5d2f5cb2e02e
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/UtlFile.java
@@ -0,0 +1,78 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/objects/UtlFile.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.objects;
+
+import org.apache.doris.hplsql.File;
+
+import java.io.EOFException;
+import java.io.IOException;
+
+public class UtlFile implements HplObject {
+ private final UtlFileClass hplClass;
+
+ public UtlFile(UtlFileClass hplClass) {
+ this.hplClass = hplClass;
+ }
+
+ @Override
+ public HplClass hplClass() {
+ return hplClass;
+ }
+
+ public File fileOpen(String dir, String name, boolean write, boolean overwrite) {
+ File file = new File();
+ if (write) {
+ file.create(dir, name, overwrite);
+ } else {
+ file.open(dir, name);
+ }
+ return file;
+ }
+
+ public void fileClose(File file) {
+ file.close();
+ }
+
+ public String getLine(File file) {
+ StringBuilder out = new StringBuilder();
+ try {
+ while (true) {
+ char c = file.readChar();
+ if (c == '\n') {
+ break;
+ }
+ out.append(c);
+ }
+ } catch (IOException e) {
+ if (!(e instanceof EOFException)) {
+ out.setLength(0);
+ }
+ }
+ return out.toString();
+ }
+
+ public void put(File file, String str, boolean newLine) {
+ file.writeString(str);
+ if (newLine) {
+ file.writeString("\n");
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/UtlFileClass.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/UtlFileClass.java
new file mode 100644
index 00000000000000..0e054e64f2555e
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/objects/UtlFileClass.java
@@ -0,0 +1,80 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/objects/UtlFileClass.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.objects;
+
+import org.apache.doris.hplsql.File;
+import org.apache.doris.hplsql.Var;
+import org.apache.doris.hplsql.objects.MethodParams.Arity;
+
+public class UtlFileClass implements HplClass {
+ public static final UtlFileClass INSTANCE = new UtlFileClass();
+ private final MethodDictionary methodDictionary = new MethodDictionary();
+
+ private UtlFileClass() {
+ methodDictionary.put("fopen", (self, args) -> {
+ MethodParams params = new MethodParams("fopen", args, Arity.min(2));
+ String dir = params.stringAt(0);
+ String name = params.stringAt(1);
+ boolean write = true;
+ boolean overwrite = false;
+ if (args.size() > 2) {
+ String mode = params.stringAt(2);
+ if (mode.equalsIgnoreCase("r")) {
+ write = false;
+ } else if (mode.equalsIgnoreCase("w")) {
+ write = true;
+ overwrite = true;
+ }
+ }
+ File file = self.fileOpen(dir, name, write, overwrite);
+ return new Var(Var.Type.FILE, file);
+ });
+ methodDictionary.put("get_line", (self, args) -> {
+ MethodParams params = new MethodParams("get_line", args, Arity.UNARY);
+ return new Var(self.getLine(params.fileAt(0)));
+ });
+ methodDictionary.put("put_line", (self, args) -> {
+ MethodParams params = new MethodParams("put_line", args, Arity.BINARY);
+ self.put(params.fileAt(0), params.stringAt(1), true);
+ return null;
+ });
+ methodDictionary.put("put", (self, args) -> {
+ MethodParams params = new MethodParams("put", args, Arity.BINARY);
+ self.put(params.fileAt(0), params.stringAt(1), false);
+ return null;
+ });
+ methodDictionary.put("fclose", (self, args) -> {
+ File file = new MethodParams("fclose", args, Arity.UNARY).fileAt(0);
+ self.fileClose(file);
+ return null;
+ });
+ }
+
+ @Override
+ public UtlFile newInstance() {
+ return new UtlFile(this);
+ }
+
+ @Override
+ public MethodDictionary methodDictionary() {
+ return methodDictionary;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/DorisPackageRegistry.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/DorisPackageRegistry.java
new file mode 100644
index 00000000000000..a5098db728df25
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/DorisPackageRegistry.java
@@ -0,0 +1,95 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/packages/HmsPackageRegistry.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.packages;
+
+import org.apache.doris.hplsql.store.HplsqlPackage;
+import org.apache.doris.hplsql.store.MetaClient;
+import org.apache.doris.qe.ConnectContext;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.thrift.TException;
+
+import java.util.Optional;
+
+public class DorisPackageRegistry implements PackageRegistry {
+ private final MetaClient client;
+
+ public DorisPackageRegistry(MetaClient client) {
+ this.client = client;
+ }
+
+ @Override
+ public Optional getPackage(String name) {
+ try {
+ HplsqlPackage pkg = findPackage(name);
+ return pkg == null
+ ? Optional.empty()
+ : Optional.of(pkg.getHeader() + ";\n" + pkg.getBody());
+ } catch (TException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void createPackageHeader(String name, String header, boolean replace) {
+ try {
+ HplsqlPackage existing = findPackage(name);
+ if (existing != null && !replace) {
+ throw new RuntimeException("Package " + name + " already exists");
+ }
+ savePackage(name, header, "");
+ } catch (TException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void createPackageBody(String name, String body, boolean replace) {
+ try {
+ HplsqlPackage existing = findPackage(name);
+ if (existing == null || StringUtils.isEmpty(existing.getHeader())) {
+ throw new RuntimeException("Package header does not exists " + name);
+ }
+ if (StringUtils.isNotEmpty(existing.getBody()) && !replace) {
+ throw new RuntimeException("Package body " + name + " already exists");
+ }
+ savePackage(name, existing.getHeader(), body);
+ } catch (TException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private HplsqlPackage findPackage(String name) throws TException {
+ return client.getHplsqlPackage(name.toUpperCase(), ConnectContext.get().getCurrentCatalog().getName(),
+ ConnectContext.get().getDatabase());
+ }
+
+ @Override
+ public void dropPackage(String name) {
+ client.dropHplsqlPackage(name, ConnectContext.get().getCurrentCatalog().getName(),
+ ConnectContext.get().getDatabase());
+ }
+
+ private void savePackage(String name, String header, String body) {
+ client.addHplsqlPackage(name.toUpperCase(), ConnectContext.get().getCurrentCatalog().getName(),
+ ConnectContext.get().getDatabase(), ConnectContext.get().getQualifiedUser(), header, body);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/InMemoryPackageRegistry.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/InMemoryPackageRegistry.java
new file mode 100644
index 00000000000000..51128bc57f587c
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/InMemoryPackageRegistry.java
@@ -0,0 +1,74 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/packages/InMemoryPackageRegistry.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.packages;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+public class InMemoryPackageRegistry implements PackageRegistry {
+ private Map registry = new HashMap<>();
+
+ @Override
+ public Optional getPackage(String name) {
+ Source src = registry.get(name.toUpperCase());
+ return src == null
+ ? Optional.empty()
+ : Optional.of(src.header + ";\n" + src.body);
+ }
+
+ @Override
+ public void createPackageHeader(String name, String header, boolean replace) {
+ if (registry.containsKey(name) && !replace) {
+ throw new RuntimeException("Package " + name + " already exits");
+ }
+ registry.put(name, new Source(header, ""));
+ }
+
+ @Override
+ public void createPackageBody(String name, String body, boolean replace) {
+ Source existing = registry.get(name);
+ if (existing == null || StringUtils.isEmpty(existing.header)) {
+ throw new RuntimeException("Package header does not exists " + name);
+ }
+ if (StringUtils.isNotEmpty(existing.body) && !replace) {
+ throw new RuntimeException("Package body " + name + " already exits");
+ }
+ registry.getOrDefault(name, new Source("", "")).body = body;
+ }
+
+ @Override
+ public void dropPackage(String name) {
+ registry.remove(name);
+ }
+
+ private static class Source {
+ String header;
+ String body;
+
+ public Source(String header, String body) {
+ this.header = header;
+ this.body = body;
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/PackageRegistry.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/PackageRegistry.java
new file mode 100644
index 00000000000000..484e41fac7363d
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/packages/PackageRegistry.java
@@ -0,0 +1,33 @@
+// 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.
+// This file is copied from
+// https://github.com/apache/hive/blob/master/hplsql/src/main/java/org/apache/hive/hplsql/packages/PackageRegistry.java
+// and modified by Doris
+
+package org.apache.doris.hplsql.packages;
+
+import java.util.Optional;
+
+public interface PackageRegistry {
+ Optional getPackage(String name);
+
+ void createPackageHeader(String name, String header, boolean replace);
+
+ void createPackageBody(String name, String body, boolean replace);
+
+ void dropPackage(String name);
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/HplsqlManager.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/HplsqlManager.java
new file mode 100644
index 00000000000000..1cc0bcf55213a4
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/HplsqlManager.java
@@ -0,0 +1,122 @@
+// 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.doris.hplsql.store;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+
+import com.google.common.collect.Maps;
+import com.google.gson.annotations.SerializedName;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Map;
+
+public class HplsqlManager implements Writable {
+ private static final Logger LOG = LogManager.getLogger(HplsqlManager.class);
+
+ @SerializedName(value = "nameToStoredProcedures")
+ Map nameToStoredProcedures = Maps.newConcurrentMap();
+
+ @SerializedName(value = "nameToPackages")
+ Map nameToPackages = Maps.newConcurrentMap();
+
+ public HplsqlManager() {
+ }
+
+ public StoredProcedure getStoredProcedure(StoredKey storedKey) {
+ return nameToStoredProcedures.get(storedKey);
+ }
+
+ public void addStoredProcedure(StoredProcedure procedure, boolean isForce) {
+ StoredKey storedKey = new StoredKey(procedure.getName(), procedure.getCatalogName(), procedure.getDbName());
+ if (isForce) {
+ nameToStoredProcedures.put(storedKey, procedure);
+ } else if (nameToStoredProcedures.putIfAbsent(storedKey, procedure) != null) {
+ throw new RuntimeException(storedKey + ", stored procedure already exist.");
+ }
+ Env.getCurrentEnv().getEditLog().logAddStoredProcedure(procedure);
+ LOG.info("Add stored procedure success: {}", storedKey);
+ }
+
+ public void replayAddStoredProcedure(StoredProcedure procedure) {
+ StoredKey storedKey = new StoredKey(procedure.getName(), procedure.getCatalogName(), procedure.getDbName());
+ nameToStoredProcedures.put(storedKey, procedure);
+ LOG.info("Replay add stored procedure success: {}", storedKey);
+ }
+
+ public void dropStoredProcedure(StoredKey storedKey) {
+ nameToStoredProcedures.remove(storedKey);
+ Env.getCurrentEnv().getEditLog().logDropStoredProcedure(storedKey);
+ LOG.info("Drop stored procedure success: {}", storedKey);
+ }
+
+ public void replayDropStoredProcedure(StoredKey storedKey) {
+ nameToStoredProcedures.remove(storedKey);
+ LOG.info("Replay drop stored procedure success: {}", storedKey);
+ }
+
+ public HplsqlPackage getPackage(StoredKey storedKey) {
+ return nameToPackages.get(storedKey);
+ }
+
+ public void addPackage(HplsqlPackage pkg, boolean isForce) {
+ StoredKey storedKey = new StoredKey(pkg.getName(), pkg.getCatalogName(), pkg.getDbName());
+ nameToPackages.put(storedKey, pkg);
+ if (isForce) {
+ nameToPackages.put(storedKey, pkg);
+ } else if (nameToPackages.putIfAbsent(storedKey, pkg) != null) {
+ throw new RuntimeException(storedKey + ", package already exist.");
+ }
+ Env.getCurrentEnv().getEditLog().logAddHplsqlPackage(pkg);
+ LOG.info("Add hplsql package success: {}", storedKey);
+ }
+
+ public void replayAddPackage(HplsqlPackage pkg) {
+ StoredKey storedKey = new StoredKey(pkg.getName(), pkg.getCatalogName(), pkg.getDbName());
+ nameToPackages.put(storedKey, pkg);
+ LOG.info("Replay add hplsql package success: {}", storedKey);
+ }
+
+ public void dropPackage(StoredKey storedKey) {
+ nameToPackages.remove(storedKey);
+ Env.getCurrentEnv().getEditLog().logDropHplsqlPackage(storedKey);
+ LOG.info("Drop hplsql package success: {}", storedKey);
+ }
+
+ public void replayDropPackage(StoredKey storedKey) {
+ nameToPackages.remove(storedKey);
+ LOG.info("Replay drop hplsql package success: {}", storedKey);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ String json = GsonUtils.GSON.toJson(this);
+ Text.writeString(out, json);
+ }
+
+ public static HplsqlManager read(DataInput in) throws IOException {
+ String json = Text.readString(in);
+ return GsonUtils.GSON.fromJson(json, HplsqlManager.class);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/HplsqlPackage.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/HplsqlPackage.java
new file mode 100644
index 00000000000000..1f84410652c7ce
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/HplsqlPackage.java
@@ -0,0 +1,74 @@
+// 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.doris.hplsql.store;
+
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.THplsqlPackage;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+@AllArgsConstructor
+@Getter
+public class HplsqlPackage implements Writable {
+ @SerializedName(value = "name")
+ private String name;
+
+ @SerializedName(value = "catalogName")
+ private String catalogName;
+
+ @SerializedName(value = "dbName")
+ private String dbName;
+
+ @SerializedName(value = "ownerName")
+ private String ownerName;
+
+ @SerializedName(value = "header")
+ private String header;
+
+ @SerializedName(value = "body")
+ private String body;
+
+ public static HplsqlPackage read(DataInput in) throws IOException {
+ String json = Text.readString(in);
+ return GsonUtils.GSON.fromJson(json, HplsqlPackage.class);
+ }
+
+ public THplsqlPackage toThrift() {
+ return new THplsqlPackage().setName(name).setCatalogName(catalogName).setDbName(dbName).setOwnerName(ownerName)
+ .setHeader(header).setBody(body);
+ }
+
+ public static HplsqlPackage fromThrift(THplsqlPackage pkg) {
+ return new HplsqlPackage(pkg.getName(), pkg.getCatalogName(), pkg.getDbName(), pkg.getOwnerName(),
+ pkg.getHeader(), pkg.getBody());
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ String json = GsonUtils.GSON.toJson(this);
+ Text.writeString(out, json);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/MetaClient.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/MetaClient.java
new file mode 100644
index 00000000000000..a2c8c97bacabe2
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/MetaClient.java
@@ -0,0 +1,198 @@
+// 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.doris.hplsql.store;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.FrontendService;
+import org.apache.doris.thrift.TAddHplsqlPackageRequest;
+import org.apache.doris.thrift.TAddStoredProcedureRequest;
+import org.apache.doris.thrift.THplsqlPackage;
+import org.apache.doris.thrift.THplsqlPackageRequest;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TStatus;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TStoredKey;
+import org.apache.doris.thrift.TStoredProcedure;
+import org.apache.doris.thrift.TStoredProcedureRequest;
+
+import org.apache.thrift.TException;
+
+import java.util.Objects;
+
+public class MetaClient {
+ public MetaClient() {
+ }
+
+ public void addStoredProcedure(String name, String catalogName, String dbName, String ownerName, String source,
+ boolean isForce) {
+ checkPriv();
+ if (Env.getCurrentEnv().isMaster()) {
+ Env.getCurrentEnv().getHplsqlManager()
+ .addStoredProcedure(new StoredProcedure(name, catalogName, dbName, ownerName, source), isForce);
+ } else {
+ addStoredProcedureThrift(name, catalogName, dbName, ownerName, source, isForce);
+ }
+ }
+
+ public void dropStoredProcedure(String name, String catalogName, String dbName) {
+ checkPriv();
+ if (Env.getCurrentEnv().isMaster()) {
+ Env.getCurrentEnv().getHplsqlManager().dropStoredProcedure(new StoredKey(name, catalogName, dbName));
+ } else {
+ dropStoredProcedureThrift(name, catalogName, dbName);
+ }
+ }
+
+ public StoredProcedure getStoredProcedure(String name, String catalogName, String dbName) {
+ return Env.getCurrentEnv().getHplsqlManager().getStoredProcedure(new StoredKey(name, catalogName, dbName));
+ }
+
+ public void addHplsqlPackage(String name, String catalogName, String dbName, String ownerName, String header,
+ String body) {
+ checkPriv();
+ if (Env.getCurrentEnv().isMaster()) {
+ Env.getCurrentEnv().getHplsqlManager()
+ .addPackage(new HplsqlPackage(name, catalogName, dbName, ownerName, header, body),
+ false);
+ } else {
+ addHplsqlPackageThrift(name, catalogName, dbName, ownerName, header, body);
+ }
+ }
+
+ public void dropHplsqlPackage(String name, String catalogName, String dbName) {
+ checkPriv();
+ if (Env.getCurrentEnv().isMaster()) {
+ Env.getCurrentEnv().getHplsqlManager().dropPackage(new StoredKey(name, catalogName, dbName));
+ } else {
+ dropHplsqlPackageThrift(name, catalogName, dbName);
+ }
+ }
+
+ public HplsqlPackage getHplsqlPackage(String name, String catalogName, String dbName) {
+ return Env.getCurrentEnv().getHplsqlManager().getPackage(new StoredKey(name, catalogName, dbName));
+ }
+
+ protected void addStoredProcedureThrift(String name, String catalogName, String dbName, String ownerName,
+ String source, boolean isForce) {
+ TStoredProcedure tStoredProcedure = new TStoredProcedure().setName(name).setCatalogName(catalogName)
+ .setDbName(dbName).setOwnerName(ownerName).setSource(source);
+ TAddStoredProcedureRequest tAddStoredProcedureRequest = new TAddStoredProcedureRequest()
+ .setStoredProcedure(tStoredProcedure);
+ tAddStoredProcedureRequest.setIsForce(isForce);
+
+ try {
+ sendUpdateRequest(tAddStoredProcedureRequest,
+ (request, client) -> client.addStoredProcedure(request).getStatus());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected void dropStoredProcedureThrift(String name, String catalogName, String dbName) {
+ TStoredKey tStoredKey = new TStoredKey().setName(name).setCatalogName(catalogName).setDbName(dbName);
+ TStoredProcedureRequest tStoredProcedureRequest = new TStoredProcedureRequest().setStoredKey(tStoredKey);
+
+ try {
+ sendUpdateRequest(tStoredProcedureRequest,
+ (request, client) -> client.dropStoredProcedure(request).getStatus());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected void addHplsqlPackageThrift(String name, String catalogName, String dbName, String ownerName,
+ String header, String body) {
+ THplsqlPackage tHplsqlPackage = new THplsqlPackage().setName(name).setCatalogName(catalogName)
+ .setDbName(dbName).setOwnerName(ownerName).setHeader(header).setBody(body);
+ TAddHplsqlPackageRequest tAddHplsqlPackageRequest = new TAddHplsqlPackageRequest()
+ .setHplsqlPackage(tHplsqlPackage);
+
+ try {
+ sendUpdateRequest(tAddHplsqlPackageRequest,
+ (request, client) -> client.addHplsqlPackage(request).getStatus());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ protected void dropHplsqlPackageThrift(String name, String catalogName, String dbName) {
+ TStoredKey tStoredKey = new TStoredKey().setName(name).setCatalogName(catalogName).setDbName(dbName);
+ THplsqlPackageRequest tHplsqlPackageRequest = new THplsqlPackageRequest().setStoredKey(tStoredKey);
+
+ try {
+ sendUpdateRequest(tHplsqlPackageRequest,
+ (request, client) -> client.dropHplsqlPackage(request).getStatus());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void checkPriv() {
+ if (!Env.getCurrentEnv().getAccessManager()
+ .checkGlobalPriv(ConnectContext.get().getCurrentUserIdentity(), PrivPredicate.ADMIN)) {
+ throw new RuntimeException(
+ "Access denied; you need (at least one of) the ADMIN privilege(s) for this operation");
+ }
+ }
+
+ private void sendUpdateRequest(Request request,
+ BiFunction sendRequest) throws Exception {
+ TNetworkAddress masterAddress = new TNetworkAddress(Env.getCurrentEnv().getMasterHost(),
+ Env.getCurrentEnv().getMasterRpcPort());
+ FrontendService.Client client = ClientPool.frontendPool.borrowObject(masterAddress);
+ TStatus status;
+ boolean isReturnToPool = true;
+ try {
+ status = sendRequest.apply(request, client);
+ checkResult(status);
+ } catch (Exception e) {
+ if (!ClientPool.frontendPool.reopen(client)) {
+ isReturnToPool = false;
+ throw e;
+ }
+
+ status = sendRequest.apply(request, client);
+ checkResult(status);
+ } finally {
+ if (isReturnToPool) {
+ ClientPool.frontendPool.returnObject(masterAddress, client);
+ } else {
+ ClientPool.frontendPool.invalidateObject(masterAddress, client);
+ }
+ }
+ }
+
+ private void checkResult(TStatus status) throws Exception {
+ if (Objects.isNull(status) || !status.isSetStatusCode()) {
+ throw new TException("Access master error, no status set.");
+ }
+ if (status.getStatusCode().equals(TStatusCode.OK)) {
+ return;
+ }
+ throw new Exception(
+ "Access fe error, code:" + status.getStatusCode().name() + ", mgs:" + status.getErrorMsgs());
+ }
+
+ @FunctionalInterface
+ public interface BiFunction {
+ R apply(T t, U u) throws Exception;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/StoredKey.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/StoredKey.java
new file mode 100644
index 00000000000000..de29a4f849babf
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/StoredKey.java
@@ -0,0 +1,90 @@
+// 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.doris.hplsql.store;
+
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.TStoredKey;
+
+import com.google.gson.annotations.SerializedName;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Objects;
+
+public class StoredKey implements Writable {
+ private static final Logger LOG = LogManager.getLogger(StoredKey.class);
+
+ @SerializedName(value = "name")
+ private String name;
+
+ @SerializedName(value = "catalogName")
+ private String catalogName;
+
+ @SerializedName(value = "dbName")
+ private String dbName;
+
+ public StoredKey(String name, String catalogName, String dbName) {
+ this.name = name;
+ this.catalogName = catalogName;
+ this.dbName = dbName;
+ }
+
+ public TStoredKey toThrift() {
+ return new TStoredKey().setName(name).setCatalogName(catalogName).setDbName(dbName);
+ }
+
+ public static StoredKey fromThrift(TStoredKey key) {
+ return new StoredKey(key.getName(), key.getCatalogName(), key.getDbName());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, catalogName, dbName);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof StoredKey)) {
+ return false;
+ }
+ return Objects.equals(this.name, ((StoredKey) obj).name) && Objects.equals(this.catalogName,
+ ((StoredKey) obj).catalogName)
+ && Objects.equals(this.dbName, ((StoredKey) obj).dbName);
+ }
+
+ @Override
+ public String toString() {
+ return "name:" + name + ", catalogName:" + catalogName + ", dbName:" + dbName;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ String json = GsonUtils.GSON.toJson(this);
+ Text.writeString(out, json);
+ }
+
+ public static StoredKey read(DataInput in) throws IOException {
+ String json = Text.readString(in);
+ return GsonUtils.GSON.fromJson(json, StoredKey.class);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/StoredProcedure.java b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/StoredProcedure.java
new file mode 100644
index 00000000000000..a13682a6cf8641
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/hplsql/store/StoredProcedure.java
@@ -0,0 +1,71 @@
+// 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.doris.hplsql.store;
+
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.thrift.TStoredProcedure;
+
+import com.google.gson.annotations.SerializedName;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+@AllArgsConstructor
+@Getter
+public class StoredProcedure implements Writable {
+ @SerializedName(value = "name")
+ private String name;
+
+ @SerializedName(value = "catalogName")
+ private String catalogName;
+
+ @SerializedName(value = "dbName")
+ private String dbName;
+
+ @SerializedName(value = "ownerName")
+ private String ownerName;
+
+ @SerializedName(value = "source")
+ private String source;
+
+ public TStoredProcedure toThrift() {
+ return new TStoredProcedure().setName(name).setCatalogName(catalogName).setDbName(dbName)
+ .setOwnerName(ownerName).setSource(source);
+ }
+
+ public static StoredProcedure fromThrift(TStoredProcedure procedure) {
+ return new StoredProcedure(procedure.getName(), procedure.getCatalogName(), procedure.getDbName(),
+ procedure.getOwnerName(), procedure.source);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ String json = GsonUtils.GSON.toJson(this);
+ Text.writeString(out, json);
+ }
+
+ public static StoredProcedure read(DataInput in) throws IOException {
+ String json = Text.readString(in);
+ return GsonUtils.GSON.fromJson(json, StoredProcedure.class);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java b/fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java
index 5f87e553ef0c83..d0c6178d958541 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java
@@ -44,6 +44,9 @@
import org.apache.doris.datasource.InitDatabaseLog;
import org.apache.doris.datasource.InitTableLog;
import org.apache.doris.ha.MasterInfo;
+import org.apache.doris.hplsql.store.HplsqlPackage;
+import org.apache.doris.hplsql.store.StoredKey;
+import org.apache.doris.hplsql.store.StoredProcedure;
import org.apache.doris.journal.bdbje.Timestamp;
import org.apache.doris.load.DeleteInfo;
import org.apache.doris.load.ExportJob;
@@ -825,6 +828,26 @@ public void readFields(DataInput in) throws IOException {
isRead = true;
break;
}
+ case OperationType.OP_ADD_STORED_PROCEDURE: {
+ data = StoredProcedure.read(in);
+ isRead = true;
+ break;
+ }
+ case OperationType.OP_DROP_STORED_PROCEDURE: {
+ data = StoredKey.read(in);
+ isRead = true;
+ break;
+ }
+ case OperationType.OP_ADD_HPLSQL_PACKAGE: {
+ data = HplsqlPackage.read(in);
+ isRead = true;
+ break;
+ }
+ case OperationType.OP_DROP_HPLSQL_PACKAGE: {
+ data = StoredKey.read(in);
+ isRead = true;
+ break;
+ }
case OperationType.OP_ALTER_DATABASE_PROPERTY: {
data = AlterDatabasePropertyInfo.read(in);
isRead = true;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapAuthenticate.java b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapAuthenticate.java
index e7644e07169c2a..f11d9813b1cc23 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapAuthenticate.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapAuthenticate.java
@@ -67,7 +67,7 @@ public static boolean authenticate(ConnectContext context, String password, Stri
UserIdentity tempUserIdentity = UserIdentity.createAnalyzedUserIdentWithIp(qualifiedUser, remoteIp);
// Search the user in doris.
List userIdentities = Env.getCurrentEnv().getAuth()
- .getUserIdentityForLdap(qualifiedUser, remoteIp);
+ .getUserIdentityUncheckPasswd(qualifiedUser, remoteIp);
UserIdentity userIdentity;
if (userIdentities.isEmpty()) {
userIdentity = tempUserIdentity;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapManager.java b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapManager.java
index 4d1404797a293e..adc79e8485bbc6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/ldap/LdapManager.java
@@ -116,7 +116,7 @@ public boolean checkUserPasswd(String fullName, String passwd) {
public boolean checkUserPasswd(String fullName, String passwd, String remoteIp, List currentUser) {
if (checkUserPasswd(fullName, passwd)) {
- currentUser.addAll(Env.getCurrentEnv().getAuth().getUserIdentityForLdap(fullName, remoteIp));
+ currentUser.addAll(Env.getCurrentEnv().getAuth().getUserIdentityUncheckPasswd(fullName, remoteIp));
return true;
}
return false;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlChannel.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlChannel.java
index 0d24314204b23b..e3b1db7eea2fed 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlChannel.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlChannel.java
@@ -461,6 +461,18 @@ public void sendOnePacket(ByteBuffer packet) throws IOException {
}
}
+ public void sendOnePacket(Object[] row) throws IOException {
+ ByteBuffer packet;
+ serializer.reset();
+ for (Object value : row) {
+ byte[] bytes = String.valueOf(value).getBytes();
+ serializer.writeVInt(bytes.length);
+ serializer.writeBytes(bytes);
+ }
+ packet = serializer.toByteBuffer();
+ sendOnePacket(packet);
+ }
+
public void sendAndFlush(ByteBuffer packet) throws IOException {
sendOnePacket(packet);
flush();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Auth.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Auth.java
index a2ab5a7b78e1f2..48204a5b75d155 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Auth.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/Auth.java
@@ -229,7 +229,7 @@ public void checkPlainPassword(String remoteUser, String remoteHost, String remo
}
}
- public List getUserIdentityForLdap(String remoteUser, String remoteHost) {
+ public List getUserIdentityUncheckPasswd(String remoteUser, String remoteHost) {
return userManager.getUserIdentityUncheckPasswd(remoteUser, remoteHost);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
index a6c7c9d91b4e8c..3e2aea51ef7230 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java
@@ -52,6 +52,9 @@
import org.apache.doris.datasource.InitCatalogLog;
import org.apache.doris.datasource.InitDatabaseLog;
import org.apache.doris.ha.MasterInfo;
+import org.apache.doris.hplsql.store.HplsqlPackage;
+import org.apache.doris.hplsql.store.StoredKey;
+import org.apache.doris.hplsql.store.StoredProcedure;
import org.apache.doris.journal.Journal;
import org.apache.doris.journal.JournalCursor;
import org.apache.doris.journal.JournalEntity;
@@ -1034,6 +1037,22 @@ public static void loadJournal(Env env, Long logId, JournalEntity journal) {
env.getAnalysisManager().replayDeleteAnalysisTask((AnalyzeDeletionLog) journal.getData());
break;
}
+ case OperationType.OP_ADD_STORED_PROCEDURE: {
+ env.getHplsqlManager().replayAddStoredProcedure((StoredProcedure) journal.getData());
+ break;
+ }
+ case OperationType.OP_DROP_STORED_PROCEDURE: {
+ env.getHplsqlManager().replayDropStoredProcedure((StoredKey) journal.getData());
+ break;
+ }
+ case OperationType.OP_ADD_HPLSQL_PACKAGE: {
+ env.getHplsqlManager().replayAddPackage((HplsqlPackage) journal.getData());
+ break;
+ }
+ case OperationType.OP_DROP_HPLSQL_PACKAGE: {
+ env.getHplsqlManager().replayDropPackage((StoredKey) journal.getData());
+ break;
+ }
case OperationType.OP_ALTER_DATABASE_PROPERTY: {
AlterDatabasePropertyInfo alterDatabasePropertyInfo = (AlterDatabasePropertyInfo) journal.getData();
LOG.info("replay alter database property: {}", alterDatabasePropertyInfo);
@@ -1590,6 +1609,22 @@ public void logDropWorkloadGroup(DropWorkloadGroupOperationLog operationLog) {
logEdit(OperationType.OP_DROP_WORKLOAD_GROUP, operationLog);
}
+ public void logAddStoredProcedure(StoredProcedure storedProcedure) {
+ logEdit(OperationType.OP_ADD_STORED_PROCEDURE, storedProcedure);
+ }
+
+ public void logDropStoredProcedure(StoredKey storedKey) {
+ logEdit(OperationType.OP_DROP_STORED_PROCEDURE, storedKey);
+ }
+
+ public void logAddHplsqlPackage(HplsqlPackage pkg) {
+ logEdit(OperationType.OP_ADD_HPLSQL_PACKAGE, pkg);
+ }
+
+ public void logDropHplsqlPackage(StoredKey storedKey) {
+ logEdit(OperationType.OP_DROP_HPLSQL_PACKAGE, storedKey);
+ }
+
public void logAlterStoragePolicy(StoragePolicy storagePolicy) {
logEdit(OperationType.OP_ALTER_STORAGE_POLICY, storagePolicy);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java b/fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java
index 2b935e74c559c2..261b86fd61b6ad 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java
@@ -313,6 +313,15 @@ public class OperationType {
// change an auto increment id for a column
public static final short OP_UPDATE_AUTO_INCREMENT_ID = 437;
+ // hplsql 440 ~ 450
+ public static final short OP_ADD_STORED_PROCEDURE = 440;
+
+ public static final short OP_DROP_STORED_PROCEDURE = 441;
+
+ public static final short OP_ADD_HPLSQL_PACKAGE = 442;
+
+ public static final short OP_DROP_HPLSQL_PACKAGE = 443;
+
/**
* Get opcode name by op code.
**/
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/meta/MetaPersistMethod.java b/fe/fe-core/src/main/java/org/apache/doris/persist/meta/MetaPersistMethod.java
index bdc3a5a2246696..dcfdc1790bdb67 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/meta/MetaPersistMethod.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/meta/MetaPersistMethod.java
@@ -230,6 +230,12 @@ public static MetaPersistMethod create(String name) throws NoSuchMethodException
metaPersistMethod.writeMethod =
Env.class.getDeclaredMethod("saveAnalysisMgr", CountingDataOutputStream.class, long.class);
break;
+ case "hplsql":
+ metaPersistMethod.readMethod = Env.class.getDeclaredMethod("loadHplsqlStored", DataInputStream.class,
+ long.class);
+ metaPersistMethod.writeMethod = Env.class.getDeclaredMethod("saveHplsqlStored",
+ CountingDataOutputStream.class, long.class);
+ break;
default:
break;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/persist/meta/PersistMetaModules.java b/fe/fe-core/src/main/java/org/apache/doris/persist/meta/PersistMetaModules.java
index 6e99a6757fe1be..e22f48aaba1a23 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/persist/meta/PersistMetaModules.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/persist/meta/PersistMetaModules.java
@@ -39,7 +39,7 @@ public class PersistMetaModules {
"globalVariable", "cluster", "broker", "resources", "exportJob", "syncJob", "backupHandler",
"paloAuth", "transactionState", "colocateTableIndex", "routineLoadJobs", "loadJobV2", "smallFiles",
"plugins", "deleteHandler", "sqlBlockRule", "policy", "mtmvJobManager", "globalFunction", "workloadGroups",
- "binlogs", "resourceGroups", "AnalysisMgr");
+ "binlogs", "resourceGroups", "AnalysisMgr", "hplsql");
// Modules in this list is deprecated and will not be saved in meta file. (also should not be in MODULE_NAMES)
public static final ImmutableList DEPRECATED_MODULE_NAMES = ImmutableList.of(
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
index dd42960ae47307..29163a891c7061 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
@@ -29,6 +29,7 @@
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.hplsql.executor.HplsqlQueryExecutor;
import org.apache.doris.mysql.DummyMysqlChannel;
import org.apache.doris.mysql.MysqlCapability;
import org.apache.doris.mysql.MysqlChannel;
@@ -148,6 +149,8 @@ public class ConnectContext {
// If set to false, the system will not restrict query resources.
private boolean isResourceTagsSet = false;
+ private HplsqlQueryExecutor hplsqlQueryExecutor = null;
+
private String sqlHash;
private JSONObject minidump = null;
@@ -269,6 +272,17 @@ public ConnectContext(StreamConnection connection) {
}
}
+ public ConnectContext createContext() {
+ ConnectContext context = new ConnectContext();
+ context.setSessionVariable(sessionVariable);
+ context.setEnv(env);
+ context.setCluster(clusterName);
+ context.setDatabase(currentDb);
+ context.setQualifiedUser(qualifiedUser);
+ context.setCurrentUserIdentity(currentUserIdentity);
+ return context;
+ }
+
public boolean isTxnModel() {
return txnEntry != null && txnEntry.isTxnModel();
}
@@ -530,6 +544,14 @@ public StmtExecutor getExecutor() {
return executor;
}
+ public HplsqlQueryExecutor getHplsqlQueryExecutor() {
+ return hplsqlQueryExecutor;
+ }
+
+ public void setHplsqlQueryExecutor(HplsqlQueryExecutor hplsqlQueryExecutor) {
+ this.hplsqlQueryExecutor = hplsqlQueryExecutor;
+ }
+
public void cleanup() {
if (mysqlChannel != null) {
mysqlChannel.close();
@@ -650,7 +672,7 @@ public void checkTimeout(long now) {
if (executor != null && executor.isInsertStmt()) {
timeoutTag = "insert";
}
- //to ms
+ // to ms
long timeout = getExecTimeout() * 1000L;
if (delta > timeout) {
LOG.warn("kill {} timeout, remote: {}, query timeout: {}",
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
index f190f68280d347..9d102dc4b9fb91 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
@@ -46,6 +46,7 @@
import org.apache.doris.common.util.SqlUtils;
import org.apache.doris.common.util.Util;
import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.hplsql.executor.HplsqlQueryExecutor;
import org.apache.doris.metric.MetricRepo;
import org.apache.doris.mysql.MysqlChannel;
import org.apache.doris.mysql.MysqlCommand;
@@ -259,7 +260,7 @@ private void handleExecute() {
ctx.setExecutor(executor);
executor.execute();
stmtStr = executeStmt.toSql();
- } catch (Throwable e) {
+ } catch (Throwable e) {
// Catch all throwable.
// If reach here, maybe palo bug.
LOG.warn("Process one query failed because unknown reason: ", e);
@@ -358,7 +359,23 @@ private void handleQuery() {
ending--;
}
String originStmt = new String(bytes, 1, ending, StandardCharsets.UTF_8);
+ try {
+ if (ctx.sessionVariable.isEnableHplsql()) {
+ HplsqlQueryExecutor hplsqlQueryExecutor = ctx.getHplsqlQueryExecutor();
+ if (hplsqlQueryExecutor == null) {
+ hplsqlQueryExecutor = new HplsqlQueryExecutor(this);
+ ctx.setHplsqlQueryExecutor(hplsqlQueryExecutor);
+ }
+ hplsqlQueryExecutor.execute(originStmt);
+ } else {
+ executeQuery(originStmt);
+ }
+ } catch (Exception ignored) {
+ //
+ }
+ }
+ public void executeQuery(String originStmt) throws Exception {
String sqlHash = DigestUtils.md5Hex(originStmt);
ctx.setSqlHash(sqlHash);
ctx.getAuditEventBuilder().reset();
@@ -453,8 +470,7 @@ private void handleQuery() {
} catch (Throwable throwable) {
handleQueryException(throwable, auditStmt, executor.getParsedStmt(),
executor.getQueryStatisticsForAuditLog());
- // execute failed, skip remaining stmts
- break;
+ throw throwable;
} finally {
executor.addProfileToSpan();
}
@@ -634,7 +650,7 @@ private ByteBuffer getResultPacket() {
// When any request is completed, it will generally need to send a response packet to the client
// This method is used to send a response packet to the client
- private void finalizeCommand() throws IOException {
+ public void finalizeCommand() throws IOException {
ByteBuffer packet;
if (executor != null && executor.isForwardToMaster()
&& ctx.getState().getStateType() != QueryState.MysqlStateType.ERR) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 3d9bddeaba91b7..80032b66b5ddf0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -334,6 +334,8 @@ public class SessionVariable implements Serializable, Writable {
// Split size for ExternalFileScanNode. Default value 0 means use the block size of HDFS/S3.
public static final String FILE_SPLIT_SIZE = "file_split_size";
+ public static final String ENABLE_HPL_SQL = "enable_hplsql";
+
/**
* use insert stmt as the unified backend for all loads
*/
@@ -991,6 +993,9 @@ public void setMaxJoinNumBushyTree(int maxJoinNumBushyTree) {
@VariableMgr.VarAttr(name = FILE_SPLIT_SIZE, needForward = true)
public long fileSplitSize = 0;
+ @VariableMgr.VarAttr(name = ENABLE_HPL_SQL)
+ public boolean enableHplSql = false;
+
/**
* determine should we enable unified load (use insert stmt as the backend for all load)
*/
@@ -1957,6 +1962,10 @@ public void checkExternalAggPartitionBits(String externalAggPartitionBits) {
}
}
+ public boolean isEnableHplsql() {
+ return enableHplSql;
+ }
+
public boolean isEnableFileCache() {
return enableFileCache;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index 19afa0ecd11a03..f2fd6cfd543e8b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -1399,7 +1399,11 @@ private void sendResult(boolean isOutfileQuery, boolean isSendFields, Queriable
}
}
-
+ if (context.sessionVariable.isEnableHplsql()) {
+ // hplsql will get the returned results without sending them to mysql client.
+ // see org/apache/doris/hplsql/executor/DorisRowResult.java
+ return;
+ }
Span fetchResultSpan = context.getTracer().spanBuilder("fetch result").setParent(Context.current()).startSpan();
try (Scope scope = fetchResultSpan.makeCurrent()) {
while (true) {
@@ -2547,6 +2551,18 @@ private List convertResultBatchToResultRows(TResultBatch batch) {
return resultRows;
}
+ public Coordinator getCoord() {
+ return coord;
+ }
+
+ public List getColumns() {
+ return parsedStmt.getColLabels();
+ }
+
+ public List getReturnTypes() {
+ return exprToType(parsedStmt.getResultExprs());
+ }
+
public SummaryProfile getSummaryProfile() {
return profile.getSummaryProfile();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index faa79efbb88586..e3750d4573b6ba 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -63,6 +63,9 @@
import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.ExternalCatalog;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.hplsql.store.HplsqlPackage;
+import org.apache.doris.hplsql.store.StoredKey;
+import org.apache.doris.hplsql.store.StoredProcedure;
import org.apache.doris.master.MasterImpl;
import org.apache.doris.mysql.privilege.AccessControllerManager;
import org.apache.doris.mysql.privilege.PrivPredicate;
@@ -87,6 +90,8 @@
import org.apache.doris.thrift.FrontendServiceVersion;
import org.apache.doris.thrift.TAddColumnsRequest;
import org.apache.doris.thrift.TAddColumnsResult;
+import org.apache.doris.thrift.TAddHplsqlPackageRequest;
+import org.apache.doris.thrift.TAddStoredProcedureRequest;
import org.apache.doris.thrift.TAutoIncrementRangeRequest;
import org.apache.doris.thrift.TAutoIncrementRangeResult;
import org.apache.doris.thrift.TBeginTxnRequest;
@@ -128,6 +133,8 @@
import org.apache.doris.thrift.TGetTablesResult;
import org.apache.doris.thrift.TGetTabletReplicaInfosRequest;
import org.apache.doris.thrift.TGetTabletReplicaInfosResult;
+import org.apache.doris.thrift.THplsqlPackageRequest;
+import org.apache.doris.thrift.THplsqlPackageResult;
import org.apache.doris.thrift.TInitExternalCtlMetaRequest;
import org.apache.doris.thrift.TInitExternalCtlMetaResult;
import org.apache.doris.thrift.TListPrivilegesResult;
@@ -165,6 +172,8 @@
import org.apache.doris.thrift.TSnapshotType;
import org.apache.doris.thrift.TStatus;
import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TStoredProcedureRequest;
+import org.apache.doris.thrift.TStoredProcedureResult;
import org.apache.doris.thrift.TStreamLoadMultiTablePutResult;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TStreamLoadPutResult;
@@ -2637,6 +2646,101 @@ private TRestoreSnapshotResult restoreSnapshotImpl(TRestoreSnapshotRequest reque
return result;
}
+ @Override
+ public TStoredProcedureResult addStoredProcedure(TAddStoredProcedureRequest request) throws TException {
+ TStoredProcedureResult result = new TStoredProcedureResult();
+ TStatus status = new TStatus(TStatusCode.OK);
+ result.setStatus(status);
+ if (!Env.getCurrentEnv().checkFeHost(getClientAddrAsString())) {
+ status.setStatusCode(TStatusCode.NOT_AUTHORIZED);
+ status.addToErrorMsgs("addStoredProcedure only accepts requests from fe.");
+ return result;
+ }
+
+ if (!request.isSetStoredProcedure()) {
+ status.setStatusCode(TStatusCode.INVALID_ARGUMENT);
+ status.addToErrorMsgs("missing stored procedure.");
+ return result;
+ }
+ try {
+ Env.getCurrentEnv().getHplsqlManager()
+ .addStoredProcedure(StoredProcedure.fromThrift(request.getStoredProcedure()),
+ request.isSetIsForce() && request.isIsForce());
+ } catch (RuntimeException e) {
+ status.setStatusCode(TStatusCode.ALREADY_EXIST);
+ status.addToErrorMsgs(e.getMessage());
+ return result;
+ }
+ return result;
+ }
+
+ @Override
+ public TStoredProcedureResult dropStoredProcedure(TStoredProcedureRequest request) throws TException {
+ TStoredProcedureResult result = new TStoredProcedureResult();
+ TStatus status = new TStatus(TStatusCode.OK);
+ result.setStatus(status);
+ if (!Env.getCurrentEnv().checkFeHost(getClientAddrAsString())) {
+ status.setStatusCode(TStatusCode.NOT_AUTHORIZED);
+ status.addToErrorMsgs("dropStoredProcedure only accepts requests from fe.");
+ return result;
+ }
+ if (!request.isSetStoredKey()) {
+ status.setStatusCode(TStatusCode.INVALID_ARGUMENT);
+ status.addToErrorMsgs("missing stored key.");
+ return result;
+ }
+
+ Env.getCurrentEnv().getHplsqlManager().dropStoredProcedure(StoredKey.fromThrift(request.getStoredKey()));
+ return result;
+ }
+
+ @Override
+ public THplsqlPackageResult addHplsqlPackage(TAddHplsqlPackageRequest request) throws TException {
+ THplsqlPackageResult result = new THplsqlPackageResult();
+ TStatus status = new TStatus(TStatusCode.OK);
+ result.setStatus(status);
+ if (!Env.getCurrentEnv().checkFeHost(getClientAddrAsString())) {
+ status.setStatusCode(TStatusCode.NOT_AUTHORIZED);
+ status.addToErrorMsgs("addHplsqlPackage only accepts requests from fe.");
+ return result;
+ }
+ if (!request.isSetHplsqlPackage()) {
+ status.setStatusCode(TStatusCode.INVALID_ARGUMENT);
+ status.addToErrorMsgs("missing hplsql package.");
+ return result;
+ }
+
+ try {
+ Env.getCurrentEnv().getHplsqlManager().addPackage(HplsqlPackage.fromThrift(request.getHplsqlPackage()),
+ request.isSetIsForce() && request.isIsForce());
+ } catch (RuntimeException e) {
+ status.setStatusCode(TStatusCode.ALREADY_EXIST);
+ status.addToErrorMsgs(e.getMessage());
+ return result;
+ }
+ return result;
+ }
+
+ @Override
+ public THplsqlPackageResult dropHplsqlPackage(THplsqlPackageRequest request) throws TException {
+ THplsqlPackageResult result = new THplsqlPackageResult();
+ TStatus status = new TStatus(TStatusCode.OK);
+ result.setStatus(status);
+ if (!Env.getCurrentEnv().checkFeHost(getClientAddrAsString())) {
+ status.setStatusCode(TStatusCode.NOT_AUTHORIZED);
+ status.addToErrorMsgs("dropHplsqlPackage only accepts requests from fe.");
+ return result;
+ }
+ if (!request.isSetStoredKey()) {
+ status.setStatusCode(TStatusCode.INVALID_ARGUMENT);
+ status.addToErrorMsgs("missing stored key.");
+ return result;
+ }
+
+ Env.getCurrentEnv().getHplsqlManager().dropPackage(StoredKey.fromThrift(request.getStoredKey()));
+ return result;
+ }
+
public TGetMasterTokenResult getMasterToken(TGetMasterTokenRequest request) throws TException {
String clientAddr = getClientAddrAsString();
LOG.debug("receive get master token request: {}", request);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapPrivsCheckerTest.java b/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapPrivsCheckerTest.java
index 54a294c8bdf1d5..e0d7463425a6f0 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapPrivsCheckerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/ldap/LdapPrivsCheckerTest.java
@@ -142,7 +142,7 @@ public void setUp() {
minTimes = 0;
result = sessionVariable;
- auth.getUserIdentityForLdap(USER, IP);
+ auth.getUserIdentityUncheckPasswd(USER, IP);
minTimes = 0;
result = Lists.newArrayList(userIdentity);
}
diff --git a/fe/pom.xml b/fe/pom.xml
index 627d95e339d8de..f9aa1df195b277 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -308,6 +308,7 @@ under the License.
6.5.1
2.0.3
1.5.4
+ 8.0.33
3.0.0
0.4.0-incubating
@@ -1402,6 +1403,17 @@ under the License.
${clickhouse.version}
all
+
+ com.oracle.database.jdbc
+ ojdbc6
+ ${ojdbc6.version}
+
+
+
+ com.mysql
+ mysql-connector-j
+ ${mysql-connector-j.version}
+
joda-time
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 06c3d96c722ee3..1ef07ad0b18516 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -1042,6 +1042,55 @@ struct TRestoreSnapshotResult {
1: optional Status.TStatus status
}
+struct TStoredProcedure {
+ 1: optional string name
+ 2: optional string catalogName
+ 3: optional string dbName
+ 4: optional string ownerName
+ 5: optional string source
+}
+
+struct THplsqlPackage {
+ 1: optional string name
+ 2: optional string catalogName
+ 3: optional string dbName
+ 4: optional string ownerName
+ 5: optional string header
+ 6: optional string body
+}
+
+struct TStoredKey {
+ 1: optional string name
+ 2: optional string catalogName
+ 3: optional string dbName
+}
+
+struct TAddStoredProcedureRequest {
+ 1: optional TStoredProcedure storedProcedure
+ 2: optional bool isForce
+}
+
+struct TStoredProcedureRequest {
+ 1: optional TStoredKey storedKey
+}
+
+struct TStoredProcedureResult {
+ 1: optional Status.TStatus status
+}
+
+struct TAddHplsqlPackageRequest {
+ 1: optional THplsqlPackage hplsqlPackage
+ 2: optional bool isForce
+}
+
+struct THplsqlPackageRequest {
+ 1: optional TStoredKey storedKey
+}
+
+struct THplsqlPackageResult {
+ 1: optional Status.TStatus status
+}
+
struct TGetMasterTokenRequest {
1: optional string cluster
2: optional string user
@@ -1140,6 +1189,11 @@ service FrontendService {
TGetTabletReplicaInfosResult getTabletReplicaInfos(1: TGetTabletReplicaInfosRequest request)
+ TStoredProcedureResult addStoredProcedure(1: TAddStoredProcedureRequest request)
+ TStoredProcedureResult dropStoredProcedure(1: TStoredProcedureRequest request)
+ THplsqlPackageResult addHplsqlPackage(1: TAddHplsqlPackageRequest request)
+ THplsqlPackageResult dropHplsqlPackage(1: THplsqlPackageRequest request)
+
TGetMasterTokenResult getMasterToken(1: TGetMasterTokenRequest request)
TGetBinlogLagResult getBinlogLag(1: TGetBinlogLagRequest request)
diff --git a/regression-test/data/hplsql_p0/test_hplsql.out b/regression-test/data/hplsql_p0/test_hplsql.out
new file mode 100644
index 00000000000000..8da53be5ba5b56
--- /dev/null
+++ b/regression-test/data/hplsql_p0/test_hplsql.out
@@ -0,0 +1,9 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+false 2 1986 1001 11011903 1243.500 false 1901-12-31 1989-03-21T13:00 wangynnsf 20.268 789.25 string12345 -170141183460469231731687303715884105727
+
+-- !select --
+1001 wangjuoo4
+1001 wangynnsf
+1002 yunlj8@nk
+
diff --git a/regression-test/suites/hplsql_p0/test_hplsql.groovy b/regression-test/suites/hplsql_p0/test_hplsql.groovy
new file mode 100644
index 00000000000000..53aeebb3a33b8d
--- /dev/null
+++ b/regression-test/suites/hplsql_p0/test_hplsql.groovy
@@ -0,0 +1,73 @@
+// 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.
+
+suite("test_hplsql") {
+ def tbl = "hplsql_tbl"
+ sql "DROP TABLE IF EXISTS ${tbl}"
+ sql """
+ create table ${tbl} (id int, name varchar(20)) DUPLICATE key(`id`) distributed by hash (`id`) buckets 4
+ properties ("replication_num"="1");
+ """
+
+ sql "set enable_hplsql = true"
+ sql "declare id INT default = 0;"
+ sql """
+ CREATE OR REPLACE PROCEDURE procedure_demo(IN name STRING, OUT result int)
+ BEGIN
+ select k1 into result from test_query_db.test where k7 = name;
+ END;
+ """
+ sql "call procedure_demo('wangynnsf', id)"
+ qt_select "select * from test_query_db.test where k1 = id"
+
+ sql """
+ CREATE OR REPLACE PROCEDURE cursor_demo()
+ BEGIN
+ DECLARE a CHAR(32);
+ DECLARE b, c INT;
+ DECLARE cur1 CURSOR FOR SELECT k7, k3 FROM test_query_db.test where k3 > 0 order by k3, k7;
+ DECLARE cur2 CURSOR FOR SELECT k4 FROM test_query_db.baseall where k4 between 0 and 21011903 order by k4;
+
+ OPEN cur1;
+ OPEN cur2;
+
+ read_loop: LOOP
+ FETCH cur1 INTO a, b;
+ IF(SQLCODE != 0) THEN
+ LEAVE read_loop;
+ END IF;
+ FETCH cur2 INTO c;
+ IF(SQLCODE != 0) THEN
+ LEAVE read_loop;
+ END IF;
+ IF b < c THEN
+ INSERT INTO ${tbl} (`name`,`id`) VALUES (a,b);
+ ELSE
+ INSERT INTO ${tbl} (`name`, `id`) VALUES (a,c);
+ END IF;
+ END LOOP;
+
+ CLOSE cur1;
+ CLOSE cur2;
+ END;
+ """
+
+ sql "call cursor_demo()"
+ qt_select """select * from ${tbl} order by 1, 2""";
+
+ sql "set enable_hplsql = false"
+}
\ No newline at end of file