cartodb-4.29/app/connectors/importer.rb

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