github-actions[bot] commented on code in PR #61141:
URL: https://github.com/apache/doris/pull/61141#discussion_r2928879923
##########
be/src/exec/sink/writer/vjdbc_table_writer.cpp:
##########
@@ -15,67 +15,100 @@
// specific language governing permissions and limitations
// under the License.
-#include "exec/sink/writer/vjdbc_table_writer.h"
+#include "vjdbc_table_writer.h"
#include <gen_cpp/DataSinks_types.h>
#include <stdint.h>
#include <sstream>
-#include "core/binary_cast.hpp"
+#include "common/logging.h"
#include "core/block/block.h"
#include "exprs/vexpr.h"
#include "exprs/vexpr_context.h"
+#include "runtime/runtime_state.h"
+#include "util/jdbc_utils.h"
namespace doris {
-JdbcConnectorParam VJdbcTableWriter::create_connect_param(const
doris::TDataSink& t_sink) {
+std::map<std::string, std::string>
VJdbcTableWriter::_build_writer_params(const TDataSink& t_sink) {
const TJdbcTableSink& t_jdbc_sink = t_sink.jdbc_table_sink;
+ std::map<std::string, std::string> params;
- JdbcConnectorParam jdbc_param;
-
- jdbc_param.catalog_id = t_jdbc_sink.jdbc_table.catalog_id;
- jdbc_param.jdbc_url = t_jdbc_sink.jdbc_table.jdbc_url;
- jdbc_param.user = t_jdbc_sink.jdbc_table.jdbc_user;
- jdbc_param.passwd = t_jdbc_sink.jdbc_table.jdbc_password;
- jdbc_param.driver_class = t_jdbc_sink.jdbc_table.jdbc_driver_class;
- jdbc_param.driver_path = t_jdbc_sink.jdbc_table.jdbc_driver_url;
- jdbc_param.driver_checksum = t_jdbc_sink.jdbc_table.jdbc_driver_checksum;
- jdbc_param.resource_name = t_jdbc_sink.jdbc_table.jdbc_resource_name;
- jdbc_param.table_type = t_jdbc_sink.table_type;
- jdbc_param.query_string = t_jdbc_sink.insert_sql;
- jdbc_param.table_name = t_jdbc_sink.jdbc_table.jdbc_table_name;
- jdbc_param.use_transaction = t_jdbc_sink.use_transaction;
- jdbc_param.connection_pool_min_size =
t_jdbc_sink.jdbc_table.connection_pool_min_size;
- jdbc_param.connection_pool_max_size =
t_jdbc_sink.jdbc_table.connection_pool_max_size;
- jdbc_param.connection_pool_max_wait_time =
t_jdbc_sink.jdbc_table.connection_pool_max_wait_time;
- jdbc_param.connection_pool_max_life_time =
t_jdbc_sink.jdbc_table.connection_pool_max_life_time;
- jdbc_param.connection_pool_keep_alive =
t_jdbc_sink.jdbc_table.connection_pool_keep_alive;
-
- return jdbc_param;
+ params["jdbc_url"] = t_jdbc_sink.jdbc_table.jdbc_url;
+ params["jdbc_user"] = t_jdbc_sink.jdbc_table.jdbc_user;
+ params["jdbc_password"] = t_jdbc_sink.jdbc_table.jdbc_password;
+ params["jdbc_driver_class"] = t_jdbc_sink.jdbc_table.jdbc_driver_class;
+ // Resolve jdbc_driver_url to absolute file:// URL
+ std::string driver_url;
+ auto resolve_st =
+
JdbcUtils::resolve_driver_url(t_jdbc_sink.jdbc_table.jdbc_driver_url,
&driver_url);
+ if (!resolve_st.ok()) {
+ LOG(WARNING) << "Failed to resolve JDBC driver URL: " <<
resolve_st.to_string();
+ driver_url = t_jdbc_sink.jdbc_table.jdbc_driver_url;
+ }
+ params["jdbc_driver_url"] = driver_url;
+
+ params["jdbc_driver_checksum"] =
t_jdbc_sink.jdbc_table.jdbc_driver_checksum;
+ params["insert_sql"] = t_jdbc_sink.insert_sql;
+ params["use_transaction"] = t_jdbc_sink.use_transaction ? "true" : "false";
+ params["catalog_id"] = std::to_string(t_jdbc_sink.jdbc_table.catalog_id);
+ params["connection_pool_min_size"] =
+ std::to_string(t_jdbc_sink.jdbc_table.connection_pool_min_size);
+ params["connection_pool_max_size"] =
+ std::to_string(t_jdbc_sink.jdbc_table.connection_pool_max_size);
+ params["connection_pool_max_wait_time"] =
+
std::to_string(t_jdbc_sink.jdbc_table.connection_pool_max_wait_time);
+ params["connection_pool_max_life_time"] =
+
std::to_string(t_jdbc_sink.jdbc_table.connection_pool_max_life_time);
+ params["connection_pool_keep_alive"] =
+ t_jdbc_sink.jdbc_table.connection_pool_keep_alive ? "true" :
"false";
+
+ return params;
}
VJdbcTableWriter::VJdbcTableWriter(const TDataSink& t_sink,
const VExprContextSPtrs& output_expr_ctxs,
std::shared_ptr<Dependency> dep,
std::shared_ptr<Dependency> fin_dep)
: AsyncResultWriter(output_expr_ctxs, dep, fin_dep),
- JdbcConnector(create_connect_param(t_sink)) {}
+ _writer_params(_build_writer_params(t_sink)),
+ _use_transaction(t_sink.jdbc_table_sink.use_transaction) {}
+
+Status VJdbcTableWriter::open(RuntimeState* state, RuntimeProfile*
operator_profile) {
+ _writer = std::make_unique<VJniFormatTransformer>(
+ state, _vec_output_expr_ctxs,
"org/apache/doris/jdbc/JdbcJniWriter", _writer_params);
+ return _writer->open();
+}
Status VJdbcTableWriter::write(RuntimeState* state, Block& block) {
Block output_block;
RETURN_IF_ERROR(_projection_block(block, &output_block));
- auto num_rows = output_block.rows();
-
- uint32_t start_send_row = 0;
- uint32_t num_row_sent = 0;
- while (start_send_row < num_rows) {
- RETURN_IF_ERROR(append(&output_block, _vec_output_expr_ctxs,
start_send_row, &num_row_sent,
- _conn_param.table_type));
- start_send_row += num_row_sent;
- num_row_sent = 0;
+
+ if (output_block.rows() == 0) {
+ return Status::OK();
+ }
+
+ return _writer->write(output_block);
+}
+
+Status VJdbcTableWriter::finish(RuntimeState* state) {
+ if (!_use_transaction || !_writer) {
+ return Status::OK();
}
+ // Call commitTrans on the Java JdbcJniWriter via JNI
+ // VJniFormatTransformer manages the JNI writer object, so we use
get_statistics
+ // to check for errors. The actual commit is done as part of close() on
the Java side
+ // when use_transaction is true.
+ // TODO: Add explicit commitTrans JNI call support to VJniFormatTransformer
+ return Status::OK();
+}
Review Comment:
**CRITICAL: Transaction commit is never called — this causes silent data
loss.**
The `finish()` method is a no-op that just returns `Status::OK()` with a
TODO comment. The old code (`VJdbcConnector::finish_trans()`) explicitly called
Java `commitTrans()` via JNI to execute `conn.commit()`. The new code path is:
1. `JdbcJniWriter.open()` sets `conn.setAutoCommit(false)` when
`useTransaction=true`
2. `writeInternal()` calls `executeBatch()` — data is written but **not
committed** (autoCommit is off)
3. `VJdbcTableWriter::finish()` — **no-op**, returns OK
4. `JdbcJniWriter.close()` — closes PreparedStatement and Connection
**without calling `conn.commit()`**
Per JDBC spec, closing a connection with `autoCommit=false` without explicit
`commit()` causes a **rollback** on most drivers (MySQL, PostgreSQL, Oracle).
HikariCP also rolls back uncommitted transactions when returning connections to
the pool.
**Result**: All transactional JDBC writes silently succeed but data is
rolled back. The operation reports success while losing all data.
**Fix**: Either:
- (a) Add a JNI call mechanism to `VJniFormatTransformer` to invoke
`commitTrans()` from `finish()`, or
- (b) Have `JdbcJniWriter.close()` call `conn.commit()` when
`useTransaction=true` before closing the connection
This must be fixed before merge.
##########
fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/JdbcJniWriter.java:
##########
@@ -0,0 +1,394 @@
+// 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.jdbc;
+
+import org.apache.doris.cloud.security.SecurityChecker;
+import org.apache.doris.common.jni.JniWriter;
+import org.apache.doris.common.jni.vec.ColumnType;
+import org.apache.doris.common.jni.vec.VectorColumn;
+import org.apache.doris.common.jni.vec.VectorTable;
+
+import com.zaxxer.hikari.HikariDataSource;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.Date;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * JdbcJniWriter writes C++ Block data to JDBC targets via PreparedStatement
batch inserts.
+ * It extends JniWriter to integrate with the unified VJniFormatTransformer
framework,
+ * following the same pattern as MaxComputeJniWriter.
+ *
+ * <p>Lifecycle (managed by C++ VJniFormatTransformer):
+ * <pre>
+ * open() -> write() [repeated] -> close()
+ * </pre>
+ *
+ * <p>Transaction control is exposed via getStatistics() responses and
+ * additional JNI method calls from C++ side.
+ *
+ * <p>Parameters (passed via constructor params map):
+ * <ul>
+ * <li>jdbc_url - JDBC connection URL</li>
+ * <li>jdbc_user - database user</li>
+ * <li>jdbc_password - database password</li>
+ * <li>jdbc_driver_class - JDBC driver class name</li>
+ * <li>jdbc_driver_url - path to driver JAR</li>
+ * <li>jdbc_driver_checksum - MD5 checksum of driver JAR</li>
+ * <li>insert_sql - INSERT SQL with ? placeholders</li>
+ * <li>use_transaction - "true"/"false"</li>
+ * <li>catalog_id - catalog ID for connection pool keying</li>
+ * <li>connection_pool_min_size - min pool size</li>
+ * <li>connection_pool_max_size - max pool size</li>
+ * <li>connection_pool_max_wait_time - max wait time (ms)</li>
+ * <li>connection_pool_max_life_time - max lifetime (ms)</li>
+ * <li>connection_pool_keep_alive - "true"/"false"</li>
+ * </ul>
+ */
+public class JdbcJniWriter extends JniWriter {
+ private static final Logger LOG = Logger.getLogger(JdbcJniWriter.class);
+
+ private final String jdbcUrl;
+ private final String jdbcUser;
+ private final String jdbcPassword;
+ private final String jdbcDriverClass;
+ private final String jdbcDriverUrl;
+ private final String jdbcDriverChecksum;
+ private final String insertSql;
+ private final boolean useTransaction;
+ private final long catalogId;
+ private final int connectionPoolMinSize;
+ private final int connectionPoolMaxSize;
+ private final int connectionPoolMaxWaitTime;
+ private final int connectionPoolMaxLifeTime;
+ private final boolean connectionPoolKeepAlive;
+
+ private HikariDataSource hikariDataSource = null;
+ private Connection conn = null;
+ private PreparedStatement preparedStatement = null;
+ private ClassLoader classLoader = null;
+
+ // Statistics
+ private long writtenRows = 0;
+ private long insertTime = 0;
+
+ public JdbcJniWriter(int batchSize, Map<String, String> params) {
+ super(batchSize, params);
+ this.jdbcUrl = params.getOrDefault("jdbc_url", "");
+ this.jdbcUser = params.getOrDefault("jdbc_user", "");
+ this.jdbcPassword = params.getOrDefault("jdbc_password", "");
+ this.jdbcDriverClass = params.getOrDefault("jdbc_driver_class", "");
+ this.jdbcDriverUrl = params.getOrDefault("jdbc_driver_url", "");
+ this.jdbcDriverChecksum = params.getOrDefault("jdbc_driver_checksum",
"");
+ this.insertSql = params.getOrDefault("insert_sql", "");
+ this.useTransaction =
"true".equalsIgnoreCase(params.getOrDefault("use_transaction", "false"));
+ this.catalogId = Long.parseLong(params.getOrDefault("catalog_id",
"0"));
+ this.connectionPoolMinSize =
Integer.parseInt(params.getOrDefault("connection_pool_min_size", "1"));
+ this.connectionPoolMaxSize =
Integer.parseInt(params.getOrDefault("connection_pool_max_size", "10"));
+ this.connectionPoolMaxWaitTime = Integer.parseInt(
+ params.getOrDefault("connection_pool_max_wait_time", "5000"));
+ this.connectionPoolMaxLifeTime = Integer.parseInt(
+ params.getOrDefault("connection_pool_max_life_time",
"1800000"));
+ this.connectionPoolKeepAlive = "true".equalsIgnoreCase(
+ params.getOrDefault("connection_pool_keep_alive", "false"));
+ }
+
+ @Override
+ public void open() throws IOException {
+ ClassLoader oldClassLoader =
Thread.currentThread().getContextClassLoader();
+ try {
+ initializeClassLoaderAndDataSource();
+
+ conn = hikariDataSource.getConnection();
+
+ if (useTransaction) {
+ conn.setAutoCommit(false);
+ }
+
+ LOG.debug("JdbcJniWriter: Preparing insert statement: " +
insertSql);
+ preparedStatement = conn.prepareStatement(insertSql);
+ } catch (Exception e) {
+ throw new IOException("JdbcJniWriter open failed: " +
e.getMessage(), e);
+ } finally {
+ Thread.currentThread().setContextClassLoader(oldClassLoader);
+ }
+ }
+
+ @Override
+ protected void writeInternal(VectorTable inputTable) throws IOException {
+ try {
+ long startInsert = System.nanoTime();
+ int numRows = inputTable.getNumRows();
+ VectorColumn[] columns = inputTable.getColumns();
+
+ for (int i = 0; i < numRows; ++i) {
+ for (int j = 0; j < columns.length; ++j) {
+ insertColumn(i, j, columns[j]);
+ }
+ preparedStatement.addBatch();
+ }
+ preparedStatement.executeBatch();
+ preparedStatement.clearBatch();
+
+ insertTime += System.nanoTime() - startInsert;
+ writtenRows += numRows;
+ } catch (SQLException e) {
+ throw new IOException("JdbcJniWriter write failed: " +
e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ try {
+ // Commit transaction if enabled (C++ side calls commitTrans via
JNI before close)
+ if (preparedStatement != null && !preparedStatement.isClosed()) {
Review Comment:
The comment says "C++ side calls commitTrans via JNI before close" but this
is **not true** — the C++ `VJdbcTableWriter::finish()` is a no-op with a TODO.
The `commitTrans()` method exists at line 198 but is never called by anyone.
This comment is misleading and masks the transaction commit bug. Either:
1. Implement the C++ side to actually call `commitTrans()` via JNI, or
2. Add `if (useTransaction && conn != null && !conn.isClosed()) {
conn.commit(); }` before closing the connection here
##########
be/src/util/jdbc_utils.cpp:
##########
@@ -0,0 +1,57 @@
+// 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.
+
+#include "util/jdbc_utils.h"
+
+#include <filesystem>
+
+#include "common/config.h"
+
+namespace doris {
+
+Status JdbcUtils::resolve_driver_url(const std::string& url, std::string*
result_url) {
+ // Already a full URL (e.g. "file:///path/to/driver.jar" or "hdfs://...")
Review Comment:
**Missing cloud-mode JDBC driver download.**
The old code (deleted `vjdbc_connector.cpp`) had cloud-mode driver download
logic. The codebase has `CloudPluginDownloader`
(`be/src/runtime/plugin/cloud_plugin_downloader.h`) which explicitly supports
`PluginType::JDBC_DRIVERS` and is tested for JDBC driver paths. However,
`resolve_driver_url()` only does local filesystem lookups.
In cloud/elastic deployments where BEs are ephemeral, the JAR won't exist
locally. The Java UDF path (`user_function_cache.cpp:551`) correctly implements
cloud-mode download:
```cpp
if (config::is_cloud_mode()) {
CloudPluginDownloader::download_from_cloud(
CloudPluginDownloader::PluginType::JAVA_UDF, url, target_path,
&downloaded_path);
}
```
Consider adding similar logic here for `PluginType::JDBC_DRIVERS` when
`config::is_cloud_mode()` is true, or document why it's not needed.
##########
be/src/format/table/jdbc_jni_reader.cpp:
##########
@@ -0,0 +1,222 @@
+// 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.
+
+#include "jdbc_jni_reader.h"
+
+#include <sstream>
+
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_string.h"
+#include "core/types.h"
+#include "exprs/function/simple_function_factory.h"
+#include "format/jni/jni_data_bridge.h"
+#include "runtime/descriptors.h"
+#include "util/jdbc_utils.h"
+
+namespace doris {
+#include "common/compile_check_begin.h"
+
+JdbcJniReader::JdbcJniReader(const std::vector<SlotDescriptor*>&
file_slot_descs,
+ RuntimeState* state, RuntimeProfile* profile,
+ const std::map<std::string, std::string>&
jdbc_params)
+ : JniReader(
+ file_slot_descs, state, profile,
"org/apache/doris/jdbc/JdbcJniScanner",
+ [&]() {
+ std::ostringstream required_fields;
+ std::ostringstream columns_types;
+ std::ostringstream replace_string;
+ int index = 0;
+ for (const auto& desc : file_slot_descs) {
+ std::string field = desc->col_name();
+ std::string type =
+
JniDataBridge::get_jni_type_with_different_string(desc->type());
+
+ // Determine replace_string for special types
+ // (bitmap, hll, quantile_state, jsonb)
+ std::string replace_type = "not_replace";
+ auto ptype = desc->type()->get_primitive_type();
+ if (ptype == PrimitiveType::TYPE_BITMAP) {
+ replace_type = "bitmap";
Review Comment:
**Bug: `TYPE_QUANTILE_STATE` is missing from the `replace_string` logic.**
The comment on line 52 says this handles "bitmap, hll, quantile_state,
jsonb", and `_is_special_type()` (line 101) correctly includes
`TYPE_QUANTILE_STATE`. However, this if-else chain only handles bitmap, hll,
and jsonb — `TYPE_QUANTILE_STATE` falls through to `"not_replace"`.
This matters because `quantile_state` is a binary-serialized type (like
bitmap and hll). On the Java side (`JdbcJniScanner.java` lines 228-264),
`"bitmap"` and `"hll"` trigger raw byte reading via `resultSet.getBytes()`,
while `"not_replace"` goes through `typeHandler.getColumnValue()` which reads
as a string. Reading binary `quantile_state` data as a string will corrupt it.
Add:
```cpp
} else if (ptype == PrimitiveType::TYPE_QUANTILE_STATE) {
replace_type = "quantile_state";
}
```
And update `JdbcJniScanner.java` to check for `"quantile_state"` alongside
`"bitmap"` and `"hll"` in the raw bytes reading path.
##########
be/src/exec/scan/jdbc_scanner.cpp:
##########
@@ -45,16 +43,48 @@ JdbcScanner::JdbcScanner(RuntimeState* state,
doris::JDBCScanLocalState* local_s
_tuple_desc(nullptr),
_table_type(table_type),
_is_tvf(is_tvf) {
- _init_profile(local_state->_scanner_profile);
_has_prepared = false;
}
+std::map<std::string, std::string> JdbcScanner::_build_jdbc_params(
+ const TupleDescriptor* tuple_desc) {
+ const JdbcTableDescriptor* jdbc_table =
+ static_cast<const JdbcTableDescriptor*>(tuple_desc->table_desc());
+
+ std::map<std::string, std::string> params;
+ params["jdbc_url"] = jdbc_table->jdbc_url();
+ params["jdbc_user"] = jdbc_table->jdbc_user();
+ params["jdbc_password"] = jdbc_table->jdbc_passwd();
+ params["jdbc_driver_class"] = jdbc_table->jdbc_driver_class();
+ // Resolve jdbc_driver_url to absolute file:// URL
+ // FE sends just the JAR filename; we need to resolve it to a full path.
+ std::string driver_url;
+ auto resolve_st =
JdbcUtils::resolve_driver_url(jdbc_table->jdbc_driver_url(), &driver_url);
+ if (!resolve_st.ok()) {
+ LOG(WARNING) << "Failed to resolve JDBC driver URL: " <<
resolve_st.to_string();
+ driver_url = jdbc_table->jdbc_driver_url();
+ }
+ params["jdbc_driver_url"] = driver_url;
+ params["query_sql"] = _query_string;
+ params["catalog_id"] = std::to_string(jdbc_table->jdbc_catalog_id());
+ params["table_type"] = _odbc_table_type_to_string(_table_type);
+ params["connection_pool_min_size"] =
std::to_string(jdbc_table->connection_pool_min_size());
+ params["connection_pool_max_size"] =
std::to_string(jdbc_table->connection_pool_max_size());
+ params["connection_pool_max_wait_time"] =
+ std::to_string(jdbc_table->connection_pool_max_wait_time());
+ params["connection_pool_max_life_time"] =
+ std::to_string(jdbc_table->connection_pool_max_life_time());
+ params["connection_pool_keep_alive"] =
+ jdbc_table->connection_pool_keep_alive() ? "true" : "false";
+ return params;
+}
Review Comment:
**Missing `jdbc_driver_checksum` parameter.**
`_build_jdbc_params()` does not include `jdbc_driver_checksum`, but the
writer path (`VJdbcTableWriter::_build_writer_params()`) does include it (line
52 of `vjdbc_table_writer.cpp`). The accessor
`jdbc_table->jdbc_driver_checksum()` is available.
The checksum is used on the Java side for driver JAR integrity validation
and cache keying. Without it, the scanner could use a stale/wrong driver JAR
from the connection pool cache.
Add:
```cpp
params["jdbc_driver_checksum"] = jdbc_table->jdbc_driver_checksum();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
