require 'uuidtools' require 'carto/importer/table_setup' require_relative '../models/visualization/support_tables' require_relative '../../lib/carto/ghost_tables_manager' require_dependency 'carto/db/user_schema' require_dependency 'visualization/derived_creator' module CartoDB module Connector class Importer ORIGIN_SCHEMA = 'cdb_importer' DESTINATION_SCHEMA = 'public' MAX_RENAME_RETRIES = 20 # The following columns are not validated because we are comparing schemas of a cartodbfied table and one that # is about to be imported. COLUMNS_NOT_TO_VALIDATE = [:cartodb_id, :the_geom_webmercator].freeze attr_reader :imported_table_visualization_ids, :rejected_layers attr_accessor :table # @param runner CartoDB::Importer2::Runner # @param table_registrar CartoDB::TableRegistrar # @param quota_checker CartoDB::QuotaChecker # @param database # @param data_import_id String UUID # @param destination_schema String|nil # @param public_user_roles Array|nil def initialize(runner:, table_registrar:, quota_checker:, database:, data_import_id:, overviews_creator: nil, destination_schema: DESTINATION_SCHEMA, public_user_roles: [CartoDB::PUBLIC_DB_USER], collision_strategy: nil) @aborted = false @runner = runner @table_registrar = table_registrar @quota_checker = quota_checker @database = database @data_import_id = data_import_id @overviews_creator = overviews_creator @destination_schema = destination_schema @support_tables_helper = CartoDB::Visualization::SupportTables.new(database, {public_user_roles: public_user_roles}) @imported_table_visualization_ids = [] @rejected_layers = [] @collision_strategy = collision_strategy @table_setup = ::Carto::Importer::TableSetup.new( user: user, overviews_creator: overviews_creator, log: runner.log ) end def run(tracker) runner.run(&tracker) if quota_checker.will_be_over_table_quota?(results.length) log('Results would set overquota') @aborted = true results.each { |result| drop(result.table_name) } else log('Checking public map quota') vis = runner.visualizations.select do |v| v.type == Carto::Visualization::TYPE_DERIVED && v.privacy != Carto::Visualization::PRIVACY_PRIVATE end if CartoDB::QuotaChecker.new(data_import.user).will_be_over_public_map_quota?(vis.count) log('Results would set public map overquota') raise CartoDB::Importer2::MapQuotaExceededError.new end log('Proceeding to register') register_results(results) results.select(&:success?).each { |result| create_overviews(result) } create_visualization if data_import.create_visualization end self end def register_results(results) Carto::GhostTablesManager.run_synchronized( user.id, attempts: 10, timeout: 3000, message: "Couldn't acquire bolt to register. Registering without bolt", user: user, import_id: @data_import_id ) do results.select(&:success?).each do |result| register(result) end end end def register(result) @support_tables_helper.reset log("Before renaming from #{result.table_name} to #{result.name}") names_taken = overwrite_strategy? ? [] : taken_names name = Carto::ValidTableNameProposer.new.propose_valid_table_name(result.name, taken_names: names_taken) overwrite = overwrite_strategy? && taken_names.include?(name) if overwrite raise ::CartoDB::Importer2::IncompatibleSchemas.new unless compatible_schemas_for_overwrite?(name) move_to_schema(result, result.table_name, ORIGIN_SCHEMA, @destination_schema) overwrite_register(result, name) do drop("\"#{@destination_schema}\".\"#{name}\"") rename(result, result.table_name, name, @destination_schema) end else new_register(name, result) end result.name = name log("Before persisting metadata '#{name}' data_import_id: #{data_import_id}") persist_metadata(name, data_import_id, overwrite) log("Table '#{name}' registered") rescue => exception if exception.message =~ /canceling statement due to statement timeout/i drop("#{ORIGIN_SCHEMA}.#{result.table_name}") raise CartoDB::Importer2::StatementTimeoutError.new( exception.message, CartoDB::Importer2::ERRORS_MAP[CartoDB::Importer2::StatementTimeoutError] ) else raise exception end end def create_overviews(result) dataset = @overviews_creator.dataset(result.name) dataset.create_overviews! rescue => exception # In case of overview creation failure we'll just omit the # overviews creation and continue with the process. # Since the actual creation is handled by a single SQL # function, and thus executed in a transaction, we shouldn't # need any clean up here. (Either all overviews were created # or nothing changed) log("Overviews creation failed: #{exception.message}") CartoDB::Logger.error( message: "Overviews creation failed", exception: exception, user: Carto::User.find(data_import.user_id), table_name: result.name ) end def create_visualization if runner.visualizations.empty? create_default_visualization else user = Carto::User.find(data_import.user_id) renamed_tables = results.map { |r| [r.original_name, r.name] }.to_h runner.visualizations.each do |visualization| persister = Carto::VisualizationsExportPersistenceService.new vis = persister.save_import(user, visualization, renamed_tables: renamed_tables) bind_visualization_to_data_import(vis) end end end def create_default_visualization tables = get_imported_tables if tables.length > 0 user = ::User.where(id: data_import.user_id).first vis, @rejected_layers = CartoDB::Visualization::DerivedCreator.new(user, tables).create bind_visualization_to_data_import(vis) end end def bind_visualization_to_data_import(vis) data_import.visualization_id = vis.id data_import.save data_import.reload end def get_imported_tables @imported_table_visualization_ids.map do |table_id| Carto::Visualization.find(table_id).table end end def success? !over_table_quota? && runner.success? end def drop_all(results) results.each { |result| drop(result.qualified_table_name) } end def drop(table_name) Carto::OverviewsService.new(database).delete_overviews table_name database.execute(%(DROP TABLE #{table_name})) rescue => exception log("Couldn't drop table #{table_name}: #{exception}. Backtrace: #{exception.backtrace} ") self end def move_to_schema(result, table_name, origin_schema, destination_schema) return self if origin_schema == destination_schema database.execute(%Q{ ALTER TABLE "#{origin_schema}"."#{table_name}" SET SCHEMA "#{destination_schema}" }) @support_tables_helper.tables = result.support_tables.map { |table| { schema: origin_schema, name: table } } @support_tables_helper.change_schema(destination_schema, table_name) rescue => e drop("#{origin_schema}.#{table_name}") raise e end # Renames table from current_name to new_name. # It doesn't check if `new_name` is valid. To get a valid name use `Carto::ValidTableNameProposer` def rename(result, current_name, new_name, schema) database.execute(%{ ALTER TABLE "#{schema}"."#{current_name}" RENAME TO "#{new_name}" }) rename_the_geom_index_if_exists(current_name, new_name, schema) @support_tables_helper.tables = result.support_tables.map { |table| { schema: schema, name: table } } # Delay recreation of constraints until schema change results = @support_tables_helper.rename(current_name, new_name, false) if results[:success] result.update_support_tables(results[:names]) else raise 'unsuccessful support tables renaming' end new_name rescue => exception drop("#{schema}.#{current_name}") CartoDB::Logger.debug(message: 'Error in table rename: dropping importer table', exception: exception, table_name: current_name, new_table_name: new_name, data_import: @data_import_id) raise exception end def rename_the_geom_index_if_exists(current_name, new_name, schema) database.execute(%Q{ ALTER INDEX IF EXISTS "#{schema}"."#{current_name}_geom_idx" RENAME TO "the_geom_#{UUIDTools::UUID.timestamp_create.to_s.gsub('-', '_')}" }) rescue => exception log("Silently failed rename_the_geom_index_if_exists from " + "#{current_name} to #{new_name} with exception #{exception}. " + "Backtrace: #{exception.backtrace}. ") end def persist_metadata(name, data_import_id, overwrite_table) user_table = overwrite_strategy? ? Carto::UserTable.where(user_id: user.id, name: name).first : nil table_registrar.register(name, data_import_id, user_table) @table = table_registrar.table @imported_table_visualization_ids << @table.table_visualization.id unless overwrite_table table.update_bounding_box self end def results runner.results end def over_table_quota? @aborted || quota_checker.over_table_quota? end def error_code return 8002 if over_table_quota? results.map(&:error_code).compact.first end def data_import @data_import ||= DataImport[@data_import_id] end def overwrite_register(result, name) index_statements = @table_setup.generate_index_statements(@destination_schema, name) database.transaction do log("Replacing #{name} with #{result.table_name}") begin # the logic inside the transaction may vary, so we let the caller to implement it yield(database, @destination_schema) rescue => e log("Unable to replace #{name} with #{result.table_name}. Rollingback transaction and dropping #{result.table_name}: #{e}") drop("\"#{@destination_schema}\".\"#{result.table_name}\"") raise e end end @table_setup.cartodbfy(name) @table_setup.fix_oid(name) @table_setup.run_index_statements(index_statements, @database) @table_setup.recreate_overviews(name) @table_setup.update_cdb_tablemetadata(name) end private def new_register(name, result) database.transaction do rename(result, result.table_name, name, ORIGIN_SCHEMA) begin log("Before moving schema '#{name}' from #{ORIGIN_SCHEMA} to #{@destination_schema}") move_to_schema(result, name, ORIGIN_SCHEMA, @destination_schema) rescue => e log("Error replacing data in import: #{e}: #{e.backtrace}") raise e end end end def compatible_schemas_for_overwrite?(name) orig_schema = user.in_database.schema(results.first.tables.first, reload: true, schema: ORIGIN_SCHEMA) dest_schema = user.in_database.schema(name, reload: true, schema: user.database_schema) dest_schema.each do |dest_row| next if COLUMNS_NOT_TO_VALIDATE.include?(dest_row[0]) return false unless orig_schema.any? { |orig_row| rows_assignable?(dest_row, orig_row) } end true end def rows_assignable?(dest_row, orig_row) (orig_row[0] == :the_geom && orig_row[0] == dest_row[0] && dest_row[1][:db_type].include?(orig_row[1][:db_type])) || (orig_row[0] == dest_row[0] && orig_row[1][:db_type].convert_to_cartodb_type == dest_row[1][:db_type].convert_to_cartodb_type) end def user table_registrar.user end def overwrite_strategy? @collision_strategy == Carto::DataImportConstants::COLLISION_STRATEGY_OVERWRITE end def log(message) runner.log.append(message) end def exists_user_table_for_user_id(table_name, user_id) !Carto::UserTable.where(name: table_name, user_id: user_id).first.nil? end def taken_names Carto::Db::UserSchema.new(table_registrar.user).table_names end attr_reader :runner, :table_registrar, :quota_checker, :database, :data_import_id end end end