Merge branch 'tickets/DM-52125'

Author: iagaponenko

doc/ingest/api/index.rst

@@ -1,7 +1,7 @@
 
 .. note::
 
-   Information in this guide corresponds to the version **39** of the Qserv REST API. Keep in mind
+   Information in this guide corresponds to the version **49** of the Qserv REST API. Keep in mind
    that each implementation of the API has a specific version. The version number will change
    if any changes to the implementation or the API that might affect users will be made.
    The current document will be kept updated to reflect the latest version of the API.

doc/ingest/api/reference/rest/controller/db-table-management.rst

@@ -46,6 +46,8 @@ that has the following schema (of which only the database and database family-re
                             "latitude_key" :            "coord_dec",
                             "create_time" :             1662774817703,
                             "unique_primary_key" :      1,
+                            "charset_name" :            "latin1",
+                            "collation_name" :          "latin1_swedish_ci",
                             "flag" :                    "",
                             "name" :                    "Source",
                             "director_database_name" :  "",
@@ -221,7 +223,9 @@ the **partitioned** tables is presented below:
         "longitude_key"        : <string>,
         "flag"                 : <string>,
         "ang_sep"              : <double>,
-        "unique_primary_key"   : <number>
+        "unique_primary_key"   : <number>,
+        "charset_name"         : <string>,
+        "collation_name"       : <string>
     }
 
 A description of the *regular* tables has a fewer number of attributes (attributes that which are specific to the *partitioned*
@@ -317,6 +321,13 @@ Where the attributes are:
   - ``0``: The primary key is not unique.
   - ``1``: The primary key is unique.
 
+``charset_name`` : *part* = ``latin1``
+  The optional parameter that affects the interpretation of the data in the CSV file when ingesting the contribution.
+  The name will be also used for setting the ``CHARSET`` attribute of the relevant MySQL tables created by the service.
+
+``collation_name`` : *part* = ``latin1_swedish_ci``
+  The optional parameter is used for setting the ``COLLATE`` attribute of the relevant MySQL tables created by the service.
+
 .. warning::
 
     - The table schema does not include definitions of indexes. Those are managed separately after the table is published.

doc/user/http-frontend-ingest.rst

@@ -79,9 +79,8 @@ Most services described in this document require user database and table names.
   This prefix is reserved for naming internal tables that Qserv places into user databases.
 
 Depending on the version of the Qserv AOI, there are additional restrictions on the names of databases and tables.
-Before version number **46** of the API, database and table names could only container alphanumeric characters and underscores.
-This restriction was imposed to avoid issues with the MySQL server. This restriction was relaxed in version **46** of the API
-to allow the following special characters:
+Before version number **46** of the API, database and table names could only contain alphanumeric characters and underscores.
+This restriction is relaxed as of version **46** of the API to allow the following special characters:
 
   -       (white space)
   - ``-`` (hyphen)
@@ -140,6 +139,8 @@ to specify the operation to be performed. The object follows this schema:
     {   "database" :        <string>,
         "table" :           <string>,
         "binary_encoding" : <string>,
+        "charset_name" :    <string>,
+        "collation_name" :  <string>,
         "timeout" :         <number>,
         "schema" :          <array>,
         "indexes" :         <array>,
@@ -159,6 +160,13 @@ Where:
 
   - :ref:`ingest-general-binary-encoding` (REST)
 
+``charset_name`` : *part* = ``latin1``
+  The optional parameter that affects the interpretation of the data in the CSV file when ingesting the contribution.
+  The name will be used for setting the ``CHARSET`` attribute of the relevant MySQL tables created by the service.
+
+``collation_name`` : *part* = ``latin1_swedish_ci``
+  The optional parameter is used for setting the ``COLLATE`` attribute of the relevant MySQL tables created by the service.
+
 ``schema`` : *array*
   The required schema definition. The schema must be a JSON array, where each entry represents a column specification.
   More information on the schema specification requirements can be found in the dedicated section of the document:
@@ -328,11 +336,11 @@ and files:
   The default value assumes the newline character.
 
 ``charset_name`` : *part* = ``latin1``
-  The optional parameters specify the desired character set name to be assumed when ingesting
-  the contribution. The default value may be also affected by the ingest services configuration.
-  See the following document for more details:
+  The optional parameter that affects the interpretation of the data in the CSV file when ingesting the contribution.
+  The name will be also used for setting the ``CHARSET`` attribute of the relevant MySQL tables created by the service.
 
-  - :ref:`ingest-api-advanced-charset` (ADVANCED)
+``collation_name`` : *part* = ``latin1_swedish_ci``
+  The optional parameter is used for setting the ``COLLATE`` attribute of the relevant MySQL tables created by the service.
 
 ``timeout`` : *part* = ``300``
   The optional timeout (in seconds) that limits the duration of the internal operations initiated by the service.
@@ -398,6 +406,8 @@ The request could be pushed to the service using:
          -F 'table=employee' \
          -F 'fields_terminated_by=,' \
          -F 'timeout=300' \
+         -F 'charset_name=utf8mb4' \
+         -F 'collation_name=utf8mb4_uca1400_ai_ci' \
          -F 'schema=@/path/to/schema.json' \
          -F 'indexes=@/path/to/indexes.json' \ 
          -F 'rows=@/path/to/employee.csv'

doc/user/http-frontend.rst

@@ -1,7 +1,7 @@
 
 .. note::
 
-  - This guide corresponds to version **47** of the Qserv REST API. Note that each API implementation has a specific version.
+  - This guide corresponds to version **49** of the Qserv REST API. Note that each API implementation has a specific version.
     The version number will change if any modifications to the implementation or API that might affect users are made.
     This document will be updated to reflect the latest API version.
   - All communications with the service are over SSL/TLS encrypted connections.

src/admin/python/lsst/qserv/admin/itest.py

@@ -1111,8 +1111,12 @@ def run_http_ingest(
     # this scenario..
     database = "user_test-db"
     table_json = "json-table"
+    table_json_utf8 = "json-table-utf8"
     table_csv = "csv$table"
+    table_csv_utf8 = "csv$table-utf8"
     timeout = 30
+    charset = "utf8mb4"
+    collation = "utf8mb4_uca1400_ai_ci"
 
     _log.debug("Testing user database: %s", database)
 
@@ -1138,6 +1142,19 @@ def run_http_ingest(
         _log.error("Failed to query table: %s of user database: %s, error: ", table_json, database, e)
         return False
 
+    # Create the table and ingest data using the JSON option. Then query the table.
+    try:
+        _http_ingest_data_json(http_frontend_uri, user, password, database, table_json_utf8, schema, indexes, rows, charset, collation)
+    except Exception as e:
+        _log.error("Failed to ingest data into table: %s of user database: %s, error: %s", table_json_utf8, database, e)
+        return False
+    try:
+        _http_query_table(http_frontend_uri, user, password, database, table_json_utf8, rows)
+    except Exception as e:
+        _log.error("Failed to query table: %s of user database: %s, error: ", table_json_utf8, database, e)
+        return False
+
+
     # Create the table and ingest data using the CSV option. Then query the table.
     try:
         _http_ingest_data_csv(http_frontend_uri, user, password, database, table_csv, schema, indexes, rows, timeout)
@@ -1149,10 +1166,21 @@ def run_http_ingest(
     except Exception as e:
         _log.error("Failed to query table: %s of user database: %s, error: ", table_csv, database, e)
 
+    # Create the table and ingest data using the CSV option. Then query the table.
+    try:
+        _http_ingest_data_csv(http_frontend_uri, user, password, database, table_csv_utf8, schema, indexes, rows, timeout, charset, collation)
+    except Exception as e:
+        _log.error("Failed to ingest data into table: %s of user database: %s, error: %s", table_csv_utf8, database, e)
+        return False
+    try:
+        _http_query_table(http_frontend_uri, user, password, database, table_csv_utf8, rows)
+    except Exception as e:
+        _log.error("Failed to query table: %s of user database: %s, error: ", table_csv_utf8, database, e)
+
     # Cleanup the tables and the database in two separate steps unless the user
     # requested to keep the results.
     if not keep_results:
-        for table in [table_json, table_csv]:
+        for table in [table_json, table_json_utf8, table_csv, table_csv_utf8]:
             try:
                 _http_delete_table(http_frontend_uri, user, password, database, table)
             except Exception as e:
@@ -1234,6 +1262,8 @@ def _http_ingest_data_json(
     schema: List[Dict[str, str]],
     indexes: List[Dict[str, Sequence[Collection[str]]]],
     rows: List[List[Any]],
+    charset: Optional[str] = None,
+    collation: Optional[str] = None,
 ) -> None:
     """Ingest data into an existing table of the user database.
 
@@ -1253,6 +1283,14 @@ def _http_ingest_data_json(
         The schema of the table to be created.
     indexes : `list` [`dict` [`str`, `list` [`list` [`str`]]]]
         The indexes of the table to be created.
+    rows : `list` [`list` [`Any`]]
+        The rows of data to be ingested into the table.
+    charset : `str`, optional
+        The character set to use for the table. If not provided, the default
+        character set will be used.
+    collation : `str`, optional
+        The collation to use for the table. If not provided, the default
+        collation will be used.
     """
     _log.debug("Ingesting JSON data into table: %s of user database: %s", table, database)
     data = {
@@ -1262,6 +1300,11 @@ def _http_ingest_data_json(
         "indexes": indexes,
         "rows": rows,
     }
+    if charset is not None:
+        data["charset_name"] = charset
+    if collation is not None:
+        data["collation_name"] = collation
+
     url = str(urljoin(http_frontend_uri, f"/ingest/data?version={repl_api_version}"))
     req = requests.post(url, json=data, verify=False, auth=(requests.auth.HTTPBasicAuth(user, password)))
     req.raise_for_status()
@@ -1279,7 +1322,9 @@ def _http_ingest_data_csv(
     schema: List[Dict[str, str]],
     indexes: List[Dict[str, Sequence[Collection[str]]]],
     rows: List[List[Any]],
-    timeout: int
+    timeout: int,
+    charset: Optional[str] = None,
+    collation: Optional[str] = None,
 ) -> None:
     """Create the table and ingest the data into the table.
 
@@ -1299,7 +1344,16 @@ def _http_ingest_data_csv(
         The schema of the table to be created.
     indexes : `list` [`dict` [`str`, `list` [`list` [`str`]]]]
         The indexes of the table to be created.
+    rows : `list` [`list` [`Any`]]
+        The rows of data to be ingested into the table.
     timeout : `int`
+        The timeout for the ingestion operation in seconds.
+    charset : `str`, optional
+        The character set to use for the table. If not provided, the default
+        character set will be used.
+    collation : `str`, optional
+        The collation to use for the table. If not provided, the default
+        collation will be used.
     """
     _log.debug("Ingesting CSV data into table: %s of user database: %s", table, database)
     base_dir = "/tmp"
@@ -1319,17 +1373,21 @@ def _http_ingest_data_csv(
         for row in rows:
             csv_writer.writerow(row)
 
-    encoder = MultipartEncoder(
-        fields = {
-            "database" : (None, database),
-            "table": (None, table),
-            "fields_terminated_by": (None, ","),
-            "timeout": (None, str(timeout)),
-            "schema": (schema_file, open(schema_file_path, "rb"), "application/json"),
-            "indexes": (indexes_file, open(indexes_file_path, "rb"), "application/json"),
-            "rows": (rows_file, open(rows_file_path, "rb"), "text/csv"),
-        }
-    )
+    fields = {
+        "database" : (None, database),
+        "table": (None, table),
+        "fields_terminated_by": (None, ","),
+        "timeout": (None, str(timeout)),
+        "schema": (schema_file, open(schema_file_path, "rb"), "application/json"),
+        "indexes": (indexes_file, open(indexes_file_path, "rb"), "application/json"),
+        "rows": (rows_file, open(rows_file_path, "rb"), "text/csv"),
+    }
+    if charset is not None:
+        fields["charset_name"] = (None, charset)
+    if collation is not None:
+        fields["collation_name"] = (None, collation)
+
+    encoder = MultipartEncoder(fields=fields)
     url = str(urljoin(http_frontend_uri, f"/ingest/csv?version={repl_api_version}"))
     req = requests.post(url, data=encoder, headers={'Content-Type': encoder.content_type}, verify=False,
                         auth=(requests.auth.HTTPBasicAuth(user, password)))

src/admin/python/lsst/qserv/admin/replicationInterface.py

@@ -41,7 +41,7 @@
 
 chunk_info_file = "chunk_info.json"
 
-repl_api_version = 48
+repl_api_version = 49
 
 _log = logging.getLogger(__name__)
 

src/czar/HttpCzarIngestCsvModule.cc

@@ -154,11 +154,12 @@ void HttpCzarIngestCsvModule::onEndOfFile() {
 
 json HttpCzarIngestCsvModule::onEndOfBody() {
     debug(__func__);
-    checkApiVersion(__func__, 39);
+    checkApiVersion(__func__, 49);
 
     _databaseName = body().required<string>("database");
     _tableName = body().required<string>("table");
     _charsetName = body().optional<string>("charset_name", "latin1");
+    _collationName = body().optional<string>("collation_name", "latin1_swedish_ci");
     _fieldsTerminatedBy = body().optional<string>("fields_terminated_by", R"(\t)");
     _fieldsEnclosedBy = body().optional<string>("fields_enclosed_by", R"(\0)");
     _fieldsEscapedBy = body().optional<string>("fields_escaped_by", R"(\\)");
@@ -178,6 +179,7 @@ json HttpCzarIngestCsvModule::onEndOfBody() {
     debug(__func__, "database: '" + _databaseName + "'");
     debug(__func__, "table: '" + _tableName + "'");
     debug(__func__, "charsetName: '" + _charsetName + "'");
+    debug(__func__, "collationName: '" + _collationName + "'");
     debug(__func__, "fields_terminated_by: '" + _fieldsTerminatedBy + "'");
     debug(__func__, "fields_enclosed_by: '" + _fieldsEnclosedBy + "'");
     debug(__func__, "fields_escaped_by: '" + _fieldsEscapedBy + "'");
@@ -216,7 +218,7 @@ json HttpCzarIngestCsvModule::onEndOfBody() {
 
     // Make changes to the persistent state of Qserv and the Replicaton/Ingest system.
     list<pair<string, string>> const warnings = ingestData(
-            _databaseName, _tableName, schema, indexes,
+            _databaseName, _tableName, _charsetName, _collationName, schema, indexes,
             [&](uint32_t transactionId) -> map<string, string> { return _pushDataToWorkers(transactionId); });
     for (auto const& warning : warnings) {
         warn(warning.first, warning.second);

src/czar/HttpCzarIngestCsvModule.h

@@ -119,6 +119,7 @@ class HttpCzarIngestCsvModule : public http::FileUploadModule, public HttpCzarIn
     std::string _databaseName;
     std::string _tableName;
     std::string _charsetName;
+    std::string _collationName;
     std::string _fieldsTerminatedBy;
     std::string _fieldsEnclosedBy;
     std::string _fieldsEscapedBy;

src/czar/HttpCzarIngestModule.cc

@@ -74,10 +74,12 @@ json HttpCzarIngestModule::executeImpl(string const& subModuleName) {
 
 json HttpCzarIngestModule::_ingestData() {
     debug(__func__);
-    checkApiVersion(__func__, 35);
+    checkApiVersion(__func__, 49);
 
     auto const databaseName = body().required<string>("database");
     auto const tableName = body().required<string>("table");
+    auto const charsetName = body().optional<string>("charset_name", "latin1");
+    auto const collationName = body().optional<string>("collation_name", "latin1_swedish_ci");
     setTimeoutSec(max(1U, body().optional<unsigned int>("timeout", timeoutSec())));
 
     // This is needed for decoding values of the binary columns should they be present
@@ -129,13 +131,15 @@ json HttpCzarIngestModule::_ingestData() {
     // Make changes to the persistent state of Qserv and the Replicaton/Ingest system.
     // Post warnings if any reported by the method.
     list<pair<string, string>> const warnings = ingestData(
-            databaseName, tableName, schema, indexes, [&](uint32_t transactionId) -> map<string, string> {
+            databaseName, tableName, charsetName, collationName, schema, indexes,
+            [&](uint32_t transactionId) -> map<string, string> {
                 // Send table data to all eligible workers and wait for the responses.
                 // Note that requests are sent in parallel, and the duration of each such request
                 // is limited by the timeout parameter.
                 json dataJson =
                         json::object({{"transaction_id", transactionId},
                                       {"table", tableName},
+                                      {"charset_name", charsetName},
                                       {"chunk", 0},
                                       {"overlap", 0},
                                       {"rows", rows},

src/czar/HttpCzarIngestModuleBase.cc

@@ -82,10 +82,11 @@ HttpCzarIngestModuleBase::HttpCzarIngestModuleBase(asio::io_service& io_service)
                            to_string(cconfig::CzarConfig::instance()->replicationRegistryPort())) {}
 
 list<pair<string, string>> HttpCzarIngestModuleBase::ingestData(
-        string const& databaseName, string const& tableName, json const& schema, json const& indexes,
+        string const& databaseName, string const& tableName, string const& charsetName,
+        string const& collationName, json const& schema, json const& indexes,
         function<map<string, string>(uint32_t)> const& submitRequestsToWorkers) {
     _unpublishOrCreateDatabase(databaseName);
-    _createTable(databaseName, tableName, schema);
+    _createTable(databaseName, tableName, charsetName, collationName, schema);
 
     uint32_t transactionId = 0;
     try {
@@ -192,9 +193,14 @@ void HttpCzarIngestModuleBase::_publishDatabase(string const& databaseName) {
 }
 
 void HttpCzarIngestModuleBase::_createTable(string const& databaseName, string const& tableName,
+                                            string const& charsetName, string const& collationName,
                                             json const& schema) {
-    json data = json::object(
-            {{"database", databaseName}, {"table", tableName}, {"is_partitioned", 0}, {"schema", schema}});
+    json data = json::object({{"database", databaseName},
+                              {"table", tableName},
+                              {"is_partitioned", 0},
+                              {"charset_name", charsetName},
+                              {"collation_name", collationName},
+                              {"schema", schema}});
     _requestController(http::Method::POST, "/ingest/table/", data);
 }