OmniSciDB  c1a53651b2
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
ForeignDataWrapperFactory.cpp
Go to the documentation of this file.
1 /*
2  * Copyright 2022 HEAVY.AI, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
18 #include "FsiJsonUtils.h"
19 
20 #include "CsvDataWrapper.h"
21 #include "ForeignDataWrapper.h"
26 #ifdef ENABLE_IMPORT_PARQUET
27 #include "ParquetDataWrapper.h"
28 #include "ParquetImporter.h"
29 #endif
30 #include "Catalog/os/UserMapping.h"
31 #include "RegexParserDataWrapper.h"
32 #include "Shared/SysDefinitions.h"
33 #include "Shared/file_path_util.h"
34 #include "Shared/misc.h"
35 #include "Shared/thread_count.h"
36 
37 namespace {
38 std::string get_data_wrapper_type(const import_export::CopyParams& copy_params) {
39  std::string data_wrapper_type;
41  data_wrapper_type = foreign_storage::DataWrapperType::CSV;
42  } else if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
44 #ifdef ENABLE_IMPORT_PARQUET
45  } else if (copy_params.source_type == import_export::SourceType::kParquetFile) {
47 #endif
48  } else {
49  UNREACHABLE();
50  }
51  return data_wrapper_type;
52 }
53 } // namespace
54 
55 namespace foreign_storage {
56 std::tuple<std::unique_ptr<foreign_storage::ForeignServer>,
57  std::unique_ptr<foreign_storage::UserMapping>,
58  std::unique_ptr<foreign_storage::ForeignTable>>
59 create_proxy_fsi_objects(const std::string& copy_from_source,
60  const import_export::CopyParams& copy_params,
61  const int db_id,
62  const TableDescriptor* table,
63  const int32_t user_id) {
65  db_id, user_id, copy_from_source, copy_params);
66 
67  CHECK(server);
68  server->validate();
69 
70  auto user_mapping =
72  db_id, user_id, copy_from_source, copy_params, server.get());
73 
74  if (user_mapping) {
75  user_mapping->validate(server.get());
76  }
77 
78  auto foreign_table =
80  db_id, table, copy_from_source, copy_params, server.get());
81 
82  CHECK(foreign_table);
83  foreign_table->validateOptionValues();
84 
85  return {std::move(server), std::move(user_mapping), std::move(foreign_table)};
86 }
87 
88 std::tuple<std::unique_ptr<foreign_storage::ForeignServer>,
89  std::unique_ptr<foreign_storage::UserMapping>,
90  std::unique_ptr<foreign_storage::ForeignTable>>
91 create_proxy_fsi_objects(const std::string& copy_from_source,
92  const import_export::CopyParams& copy_params,
93  const TableDescriptor* table) {
94  return create_proxy_fsi_objects(copy_from_source, copy_params, -1, table, -1);
95 }
96 
97 } // namespace foreign_storage
98 
99 namespace {
100 
101 bool is_valid_data_wrapper(const std::string& data_wrapper_type) {
102  return
103 #ifdef ENABLE_IMPORT_PARQUET
104  data_wrapper_type == foreign_storage::DataWrapperType::PARQUET ||
105 #endif
106  data_wrapper_type == foreign_storage::DataWrapperType::CSV ||
108 }
109 
110 } // namespace
111 
112 namespace foreign_storage {
113 
115  if (copy_params.line_regex.empty()) {
116  throw std::runtime_error{"Regex parser options must contain a line regex."};
117  }
118 }
119 
121  return
122 #ifdef ENABLE_IMPORT_PARQUET
124 #endif
127 }
128 
129 std::string bool_to_option_value(const bool value) {
130  return value ? "TRUE" : "FALSE";
131 }
132 
133 std::unique_ptr<ForeignDataWrapper> ForeignDataWrapperFactory::createForGeneralImport(
134  const import_export::CopyParams& copy_params,
135  const int db_id,
136  const ForeignTable* foreign_table,
137  const UserMapping* user_mapping) {
138  auto data_wrapper_type = get_data_wrapper_type(copy_params);
139  CHECK(is_valid_data_wrapper(data_wrapper_type));
140 
141  if (data_wrapper_type == DataWrapperType::CSV) {
142  return std::make_unique<CsvDataWrapper>(
143  db_id, foreign_table, user_mapping, /*disable_cache=*/true);
144  } else if (data_wrapper_type == DataWrapperType::REGEX_PARSER) {
145  return std::make_unique<RegexParserDataWrapper>(
146  db_id, foreign_table, user_mapping, true);
147  }
148 #ifdef ENABLE_IMPORT_PARQUET
149  else if (data_wrapper_type == DataWrapperType::PARQUET) {
150  return std::make_unique<ParquetDataWrapper>(
151  db_id, foreign_table, /*do_metadata_stats_validation=*/false);
152  }
153 #endif
154 
155  return {};
156 }
157 
158 std::unique_ptr<ForeignDataWrapper> ForeignDataWrapperFactory::createForImport(
159  const std::string& data_wrapper_type,
160  const int db_id,
161  const ForeignTable* foreign_table,
162  const UserMapping* user_mapping) {
163 #ifdef ENABLE_IMPORT_PARQUET
164  // only supported for parquet import path currently
165  CHECK(data_wrapper_type == DataWrapperType::PARQUET);
166  return std::make_unique<ParquetImporter>(db_id, foreign_table, user_mapping);
167 #else
168  return {};
169 #endif
170 }
171 
172 std::unique_ptr<UserMapping>
174  const int db_id,
175  const int user_id,
176  const std::string& file_path,
177  const import_export::CopyParams& copy_params,
178  const ForeignServer* server) {
179  return {};
180 }
181 
183  const int db_id,
184  const int user_id,
185  const std::string& file_path,
186  const import_export::CopyParams& copy_params) {
187  CHECK(is_valid_source_type(copy_params));
188 
189  auto foreign_server = std::make_unique<foreign_storage::ForeignServer>();
190 
191  foreign_server->id = -1;
192  foreign_server->user_id = user_id;
194  foreign_server->data_wrapper_type = DataWrapperType::CSV;
195  } else if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
196  foreign_server->data_wrapper_type = DataWrapperType::REGEX_PARSER;
197 #ifdef ENABLE_IMPORT_PARQUET
198  } else if (copy_params.source_type == import_export::SourceType::kParquetFile) {
199  foreign_server->data_wrapper_type = DataWrapperType::PARQUET;
200 #endif
201  } else {
202  UNREACHABLE();
203  }
204  foreign_server->name = "import_proxy_server";
205 
206  if (copy_params.source_type == import_export::SourceType::kOdbc) {
207  throw std::runtime_error("ODBC storage not supported");
208  } else if (shared::is_s3_uri(file_path)) {
209  throw std::runtime_error("AWS storage not supported");
210  } else {
211  foreign_server->options[AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY] =
213  }
214 
215  return foreign_server;
216 }
217 
218 namespace {
220  const import_export::ImportHeaderRow& has_header) {
221  switch (has_header) {
223  options[CsvFileBufferParser::HEADER_KEY] = "FALSE";
224  break;
227  options[CsvFileBufferParser::HEADER_KEY] = "TRUE";
228  break;
229  default:
230  CHECK(false);
231  }
232 }
233 } // namespace
234 
236  const int db_id,
237  const TableDescriptor* table,
238  const std::string& copy_from_source,
239  const import_export::CopyParams& copy_params,
240  const ForeignServer* server) {
241  CHECK(is_valid_source_type(copy_params));
242 
243  auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id);
244  auto foreign_table = std::make_unique<ForeignTable>();
245 
246  *static_cast<TableDescriptor*>(foreign_table.get()) =
247  *table; // copy table related values
248 
249  CHECK(server);
250  foreign_table->foreign_server = server;
251 
252  // populate options for regex filtering of file-paths in supported data types
256  if (copy_params.regex_path_filter.has_value()) {
258  copy_params.regex_path_filter.value();
259  }
260  if (copy_params.file_sort_order_by.has_value()) {
262  copy_params.file_sort_order_by.value();
263  }
264  if (copy_params.file_sort_regex.has_value()) {
266  copy_params.file_sort_regex.value();
267  }
268  foreign_table->options[AbstractFileStorageDataWrapper::THREADS_KEY] =
270  }
271 
273  CHECK(!copy_params.line_regex.empty());
274  foreign_table->options[RegexFileBufferParser::LINE_REGEX_KEY] =
275  copy_params.line_regex;
276  if (!copy_params.line_start_regex.empty()) {
277  foreign_table->options[RegexFileBufferParser::LINE_START_REGEX_KEY] =
278  copy_params.line_start_regex;
279  }
281  set_header_option(foreign_table->options, copy_params.has_header);
282  }
283  }
284 
285  // setup data source options based on various criteria
286  if (copy_params.source_type == import_export::SourceType::kOdbc) {
287  throw std::runtime_error("ODBC storage not supported");
288  } else if (shared::is_s3_uri(copy_from_source)) {
289  throw std::runtime_error("AWS storage not supported");
290  } else {
291  foreign_table->options["FILE_PATH"] = copy_from_source;
292  }
293 
294  // for CSV import
296  foreign_table->options[CsvFileBufferParser::DELIMITER_KEY] = copy_params.delimiter;
297  foreign_table->options[CsvFileBufferParser::NULLS_KEY] = copy_params.null_str;
298  set_header_option(foreign_table->options, copy_params.has_header);
299  foreign_table->options[CsvFileBufferParser::QUOTED_KEY] =
300  bool_to_option_value(copy_params.quoted);
301  foreign_table->options[CsvFileBufferParser::QUOTE_KEY] = copy_params.quote;
302  foreign_table->options[CsvFileBufferParser::ESCAPE_KEY] = copy_params.escape;
303  foreign_table->options[CsvFileBufferParser::LINE_DELIMITER_KEY] =
304  copy_params.line_delim;
305  foreign_table->options[CsvFileBufferParser::ARRAY_DELIMITER_KEY] =
306  copy_params.array_delim;
307  const std::array<char, 3> array_marker{
308  copy_params.array_begin, copy_params.array_end, 0};
309  foreign_table->options[CsvFileBufferParser::ARRAY_MARKER_KEY] = array_marker.data();
310  foreign_table->options[CsvFileBufferParser::LONLAT_KEY] =
311  bool_to_option_value(copy_params.lonlat);
312  foreign_table->options[CsvFileBufferParser::GEO_ASSIGN_RENDER_GROUPS_KEY] =
314  if (copy_params.geo_explode_collections) {
315  throw std::runtime_error(
316  "geo_explode_collections is not yet supported for FSI CSV import");
317  }
318  foreign_table->options[CsvFileBufferParser::GEO_EXPLODE_COLLECTIONS_KEY] =
320  foreign_table->options[CsvFileBufferParser::SOURCE_SRID_KEY] =
321  std::to_string(copy_params.source_srid);
322 
323  foreign_table->options[TextFileBufferParser::BUFFER_SIZE_KEY] =
324  std::to_string(copy_params.buffer_size);
325 
326  foreign_table->options[CsvFileBufferParser::TRIM_SPACES_KEY] =
327  bool_to_option_value(copy_params.trim_spaces);
328  }
329 
330  foreign_table->initializeOptions();
331  return foreign_table;
332 }
333 
334 std::unique_ptr<ForeignDataWrapper> ForeignDataWrapperFactory::create(
335  const std::string& data_wrapper_type,
336  const int db_id,
337  const ForeignTable* foreign_table) {
338  std::unique_ptr<ForeignDataWrapper> data_wrapper;
339  if (data_wrapper_type == DataWrapperType::CSV) {
340  if (CsvDataWrapper::validateAndGetIsS3Select(foreign_table)) {
341  UNREACHABLE();
342  } else {
343  data_wrapper = std::make_unique<CsvDataWrapper>(db_id, foreign_table);
344  }
345 #ifdef ENABLE_IMPORT_PARQUET
346  } else if (data_wrapper_type == DataWrapperType::PARQUET) {
347  data_wrapper = std::make_unique<ParquetDataWrapper>(db_id, foreign_table);
348 #endif
349  } else if (data_wrapper_type == DataWrapperType::REGEX_PARSER) {
350  data_wrapper = std::make_unique<RegexParserDataWrapper>(db_id, foreign_table);
351  } else if (data_wrapper_type == DataWrapperType::INTERNAL_CATALOG) {
352  data_wrapper = std::make_unique<InternalCatalogDataWrapper>(db_id, foreign_table);
353  } else if (data_wrapper_type == DataWrapperType::INTERNAL_MEMORY_STATS) {
354  data_wrapper = std::make_unique<InternalMemoryStatsDataWrapper>(db_id, foreign_table);
355  } else if (data_wrapper_type == DataWrapperType::INTERNAL_STORAGE_STATS) {
356  data_wrapper =
357  std::make_unique<InternalStorageStatsDataWrapper>(db_id, foreign_table);
358  } else if (data_wrapper_type == DataWrapperType::INTERNAL_LOGS) {
359  data_wrapper = std::make_unique<InternalLogsDataWrapper>(db_id, foreign_table);
360  } else {
361  throw std::runtime_error("Unsupported data wrapper");
362  }
363  return data_wrapper;
364 }
365 
367  const std::string& data_wrapper_type,
368  const ForeignTable* foreign_table) {
369  bool is_s3_select_wrapper{false};
370  std::string data_wrapper_type_key{data_wrapper_type};
371  constexpr const char* S3_SELECT_WRAPPER_KEY = "CSV_S3_SELECT";
372  if (foreign_table && data_wrapper_type == DataWrapperType::CSV &&
374  is_s3_select_wrapper = true;
375  data_wrapper_type_key = S3_SELECT_WRAPPER_KEY;
376  }
377 
378  if (validation_data_wrappers_.find(data_wrapper_type_key) ==
380  if (data_wrapper_type == DataWrapperType::CSV) {
381  if (is_s3_select_wrapper) {
382  UNREACHABLE();
383  } else {
384  validation_data_wrappers_[data_wrapper_type_key] =
385  std::make_unique<CsvDataWrapper>();
386  }
387 #ifdef ENABLE_IMPORT_PARQUET
388  } else if (data_wrapper_type == DataWrapperType::PARQUET) {
389  validation_data_wrappers_[data_wrapper_type_key] =
390  std::make_unique<ParquetDataWrapper>();
391 #endif
392  } else if (data_wrapper_type == DataWrapperType::REGEX_PARSER) {
393  validation_data_wrappers_[data_wrapper_type_key] =
394  std::make_unique<RegexParserDataWrapper>();
395  } else if (data_wrapper_type == DataWrapperType::INTERNAL_CATALOG) {
396  validation_data_wrappers_[data_wrapper_type_key] =
397  std::make_unique<InternalCatalogDataWrapper>();
398  } else if (data_wrapper_type == DataWrapperType::INTERNAL_MEMORY_STATS) {
399  validation_data_wrappers_[data_wrapper_type_key] =
400  std::make_unique<InternalMemoryStatsDataWrapper>();
401  } else if (data_wrapper_type == DataWrapperType::INTERNAL_STORAGE_STATS) {
402  validation_data_wrappers_[data_wrapper_type_key] =
403  std::make_unique<InternalStorageStatsDataWrapper>();
404  } else if (data_wrapper_type == DataWrapperType::INTERNAL_LOGS) {
405  validation_data_wrappers_[data_wrapper_type_key] =
406  std::make_unique<InternalLogsDataWrapper>();
407  } else {
408  UNREACHABLE();
409  }
410  }
411  CHECK(validation_data_wrappers_.find(data_wrapper_type_key) !=
413  return *validation_data_wrappers_[data_wrapper_type_key];
414 }
415 
417  const std::string& data_wrapper_type) {
418  const auto& supported_wrapper_types = DataWrapperType::supported_data_wrapper_types;
419  if (std::find(supported_wrapper_types.begin(),
420  supported_wrapper_types.end(),
421  data_wrapper_type) == supported_wrapper_types.end()) {
422  std::vector<std::string_view> user_facing_wrapper_types;
423  for (const auto& type : supported_wrapper_types) {
425  user_facing_wrapper_types.emplace_back(type);
426  }
427  }
428  throw std::runtime_error{"Invalid data wrapper type \"" + data_wrapper_type +
429  "\". Data wrapper type must be one of the following: " +
430  join(user_facing_wrapper_types, ", ") + "."};
431  }
432 }
433 
434 std::map<std::string, std::unique_ptr<ForeignDataWrapper>>
436 } // namespace foreign_storage
bool contains(const T &container, const U &element)
Definition: misc.h:195
static const std::string GEO_EXPLODE_COLLECTIONS_KEY
static const ForeignDataWrapper & createForValidation(const std::string &data_wrapper_type, const ForeignTable *foreign_table=nullptr)
bool is_valid_data_wrapper(const std::string &data_wrapper_type)
static std::unique_ptr< ForeignDataWrapper > createForImport(const std::string &data_wrapper_type, const int db_id, const ForeignTable *foreign_table, const UserMapping *user_mapping)
static constexpr char const * REGEX_PARSER
static std::unique_ptr< ForeignDataWrapper > create(const std::string &data_wrapper_type, const int db_id, const ForeignTable *foreign_table)
shared utility for globbing files, paths can be specified as either a single file, directory or wildcards
static const std::string TRIM_SPACES_KEY
static const std::string ARRAY_MARKER_KEY
bool is_s3_uri(const std::string &file_path)
static constexpr char const * INTERNAL_STORAGE_STATS
static constexpr std::array< std::string_view, 7 > supported_data_wrapper_types
std::string join(T const &container, std::string const &delim)
static constexpr std::array< char const *, 4 > INTERNAL_DATA_WRAPPERS
#define UNREACHABLE()
Definition: Logger.h:337
static bool validateAndGetIsS3Select(const ForeignTable *foreign_table)
void set_header_option(OptionsMap &options, const import_export::ImportHeaderRow &has_header)
static constexpr char const * INTERNAL_CATALOG
std::string to_string(char const *&&v)
std::tuple< std::unique_ptr< foreign_storage::ForeignServer >, std::unique_ptr< foreign_storage::UserMapping >, std::unique_ptr< foreign_storage::ForeignTable > > create_proxy_fsi_objects(const std::string &copy_from_source, const import_export::CopyParams &copy_params, const int db_id, const TableDescriptor *table, const int32_t user_id)
Create proxy fsi objects for use outside FSI.
static std::map< std::string, std::unique_ptr< ForeignDataWrapper > > validation_data_wrappers_
ImportHeaderRow has_header
Definition: CopyParams.h:46
static const std::string SOURCE_SRID_KEY
std::optional< std::string > regex_path_filter
Definition: CopyParams.h:85
static std::unique_ptr< ForeignDataWrapper > createForGeneralImport(const import_export::CopyParams &copy_params, const int db_id, const ForeignTable *foreign_table, const UserMapping *user_mapping)
static SysCatalog & instance()
Definition: SysCatalog.h:343
std::string bool_to_option_value(const bool value)
static void validateDataWrapperType(const std::string &data_wrapper_type)
std::string get_data_wrapper_type(const import_export::CopyParams &copy_params)
static const std::string LINE_DELIMITER_KEY
import_export::SourceType source_type
Definition: CopyParams.h:57
bool is_valid_source_type(const import_export::CopyParams &copy_params)
size_t num_import_threads(const int32_t copy_params_threads)
Definition: thread_count.h:31
static const std::string DELIMITER_KEY
std::shared_ptr< Catalog > getCatalog(const std::string &dbName)
void validate_regex_parser_options(const import_export::CopyParams &copy_params)
static const std::string ARRAY_DELIMITER_KEY
std::string line_start_regex
Definition: CopyParams.h:106
static constexpr char const * INTERNAL_MEMORY_STATS
static std::unique_ptr< ForeignServer > createForeignServerProxy(const int db_id, const int user_id, const std::string &file_path, const import_export::CopyParams &copy_params)
static std::unique_ptr< ForeignTable > createForeignTableProxy(const int db_id, const TableDescriptor *table, const std::string &file_path, const import_export::CopyParams &copy_params, const ForeignServer *server)
#define CHECK(condition)
Definition: Logger.h:291
static constexpr char const * CSV
std::map< std::string, std::string, std::less<>> OptionsMap
static const std::string GEO_ASSIGN_RENDER_GROUPS_KEY
static constexpr char const * INTERNAL_LOGS
std::optional< std::string > file_sort_order_by
Definition: CopyParams.h:86
static constexpr char const * PARQUET
std::optional< std::string > file_sort_regex
Definition: CopyParams.h:87
static std::unique_ptr< UserMapping > createUserMappingProxyIfApplicable(const int db_id, const int user_id, const std::string &file_path, const import_export::CopyParams &copy_params, const ForeignServer *server)