Initial commit
This commit is contained in:
zhongjin
@@ -0,0 +1,423 @@
|
||||
require 'carto/importer/table_setup'
|
||||
|
||||
module CartoDB
|
||||
module Synchronization
|
||||
class Adapter
|
||||
STATEMENT_TIMEOUT = (1.hour * 1000).freeze
|
||||
DESTINATION_SCHEMA = 'public'.freeze
|
||||
THE_GEOM = 'the_geom'.freeze
|
||||
OVERWRITE_ERROR = 2013
|
||||
|
||||
def initialize(table_name, runner, database, user, overviews_creator, synchronization_id)
|
||||
@table_name = table_name
|
||||
@runner = runner
|
||||
@database = database
|
||||
@user = user
|
||||
@overviews_creator = overviews_creator
|
||||
@failed = false
|
||||
@table_setup = ::Carto::Importer::TableSetup.new(
|
||||
user: user,
|
||||
overviews_creator: overviews_creator,
|
||||
log: runner.log
|
||||
)
|
||||
@error_code = nil
|
||||
@synchronization_id = synchronization_id
|
||||
end
|
||||
|
||||
def run(&tracker)
|
||||
runner.run(&tracker)
|
||||
result = runner.results.select(&:success?).first
|
||||
|
||||
if runner.remote_data_updated?
|
||||
if result.nil?
|
||||
data_for_exception = "Expecting success data for table '#{table_name}'\nResults:#{runner.results.to_s}\n"
|
||||
data_for_exception << "1st result:#{runner.results.first.inspect}"
|
||||
raise data_for_exception
|
||||
end
|
||||
|
||||
Carto::GhostTablesManager.run_synchronized(
|
||||
user.id, attempts: 10, timeout: 3000,
|
||||
message: "Couldn't acquire bolt to register. Registering sync without bolt",
|
||||
user: user,
|
||||
synchronization_id: @synchronization_id
|
||||
) do
|
||||
move_to_schema(result)
|
||||
geo_type = fix_the_geom_type!(user.database_schema, result.table_name)
|
||||
import_cleanup(user.database_schema, result.table_name)
|
||||
@table_setup.cartodbfy(result.table_name)
|
||||
overwrite(user.database_schema, table_name, result, geo_type)
|
||||
setup_table(table_name, geo_type)
|
||||
@table_setup.recreate_overviews(table_name)
|
||||
end
|
||||
end
|
||||
self
|
||||
rescue => exception
|
||||
@failed = true
|
||||
puts '=================='
|
||||
puts exception.to_s
|
||||
puts exception.backtrace
|
||||
puts '=================='
|
||||
drop(result.table_name) if result && exists?(result.table_name)
|
||||
raise exception
|
||||
end
|
||||
|
||||
def user
|
||||
@user
|
||||
end
|
||||
|
||||
def overwrite(schema, table_name, result, geo_type)
|
||||
# Determine what kind of overwrite to perform
|
||||
# overwrite_replace substitutes the existing table by the new one,
|
||||
# so any modifications since previous import/sync are lost.
|
||||
# overwrite_sync will preserve columns added since the import/sync,
|
||||
# and the geometry as well if the new table doesn't have it (nil geo_type)
|
||||
|
||||
# For the time being the latter method will only be used with tables
|
||||
# that have had the geocoder analysis applied, resulting in an column
|
||||
# named carto_geocode_hash being present.
|
||||
# TODO: we could perform the sync if there's any column named `_carto_*`
|
||||
# (carto_geocode_hash would need be renamed as _carto_geocode_hash)
|
||||
sync = has_column(schema, table_name, 'carto_geocode_hash')
|
||||
|
||||
if sync
|
||||
overwrite_sync(schema, table_name, result, geo_type)
|
||||
else
|
||||
overwrite_replace(schema, table_name, result)
|
||||
end
|
||||
end
|
||||
|
||||
def overwrite_sync(schema, table_name, result, geo_type)
|
||||
return false unless runner.remote_data_updated?
|
||||
|
||||
# NOTE the import table is already moved to the user schema;
|
||||
# this was done (#7543) because the cartodbfication performs
|
||||
# queries on CDB_UserQuotaSize and other functions expected
|
||||
# to exist in the schema of the table.
|
||||
qualified_result_table_name = %{"#{schema}"."#{result.table_name}"}
|
||||
skip_columns = '{the_geom, the_geom_webmercator}'
|
||||
|
||||
database.transaction do
|
||||
if geo_type.nil?
|
||||
# If there's no geometry in the result table, not worth
|
||||
# syncing. Maybe those were added via geocoding
|
||||
database.execute(%{
|
||||
SELECT cartodb.CDB_SyncTable(
|
||||
'#{qualified_result_table_name}',
|
||||
'#{schema}', '#{table_name}',
|
||||
'#{skip_columns}'
|
||||
)})
|
||||
else
|
||||
database.execute(%{
|
||||
SELECT cartodb.CDB_SyncTable(
|
||||
'#{qualified_result_table_name}',
|
||||
'#{schema}', '#{table_name}'
|
||||
)})
|
||||
end
|
||||
end
|
||||
|
||||
drop(result.table_name) if exists?(result.table_name)
|
||||
|
||||
# TODO not sure whether these two are needed
|
||||
@table_setup.fix_oid(table_name)
|
||||
@table_setup.update_cdb_tablemetadata(table_name)
|
||||
rescue => exception
|
||||
@error_code = OVERWRITE_ERROR
|
||||
puts "Sync overwrite ERROR: #{exception.message}: #{exception.backtrace.join}"
|
||||
|
||||
# Gets all attributes in the result except for 'log_trace', as it is too long for Rollbar
|
||||
result_hash = CartoDB::Importer2::Result::ATTRIBUTES.map { |m| [m, result.send(m)] if m != 'log_trace' }
|
||||
.compact.to_h
|
||||
CartoDB::Logger.error(message: 'Error in sync overwrite',
|
||||
exception: exception,
|
||||
user: user,
|
||||
table: table_name,
|
||||
result: result_hash)
|
||||
drop(result.table_name) if exists?(result.table_name)
|
||||
raise exception
|
||||
end
|
||||
|
||||
def overwrite_replace(schema, table_name, result)
|
||||
return false unless runner.remote_data_updated?
|
||||
|
||||
@table_setup.copy_privileges(schema, table_name, schema, result.table_name)
|
||||
index_statements = @table_setup.generate_index_statements(schema, table_name)
|
||||
|
||||
temporary_name = temporary_name_for(result.table_name)
|
||||
database.transaction do
|
||||
rename(table_name, temporary_name) if exists?(table_name)
|
||||
drop(temporary_name) if exists?(temporary_name)
|
||||
rename(result.table_name, table_name)
|
||||
end
|
||||
@table_setup.fix_oid(table_name)
|
||||
@table_setup.update_cdb_tablemetadata(table_name)
|
||||
@table_setup.run_index_statements(index_statements, @database)
|
||||
rescue => exception
|
||||
@error_code = OVERWRITE_ERROR
|
||||
puts "Sync overwrite ERROR: #{exception.message}: #{exception.backtrace.join}"
|
||||
|
||||
# Gets all attributes in the result except for 'log_trace', as it is too long for Rollbar
|
||||
result_hash = CartoDB::Importer2::Result::ATTRIBUTES.map { |m| [m, result.send(m)] if m != 'log_trace' }
|
||||
.compact.to_h
|
||||
CartoDB::Logger.error(message: 'Error in sync overwrite',
|
||||
exception: exception,
|
||||
user: user,
|
||||
table: table_name,
|
||||
result: result_hash)
|
||||
drop(result.table_name) if exists?(result.table_name)
|
||||
raise exception
|
||||
end
|
||||
|
||||
def setup_table(table_name, geo_type)
|
||||
table = Carto::UserTable.find(user.tables.where(name: table_name).first.id).service
|
||||
|
||||
table.force_schema = true
|
||||
|
||||
table.import_to_cartodb(table_name)
|
||||
table.schema(reload: true)
|
||||
table.reload
|
||||
|
||||
# We send the detected geometry type to avoid manipulating geoms twice
|
||||
# set_the_geom_column! should just edit the metadata with the specified type
|
||||
table.send :set_the_geom_column!, geo_type
|
||||
table.save
|
||||
rescue => exception
|
||||
CartoDB::Logger.error(message: 'Error in setup cartodbfy',
|
||||
exception: exception,
|
||||
user: user,
|
||||
table: table_name)
|
||||
ensure
|
||||
@table_setup.fix_oid(table_name)
|
||||
end
|
||||
|
||||
def has_column(schema_name, table_name, column_name)
|
||||
qualified_table_name = "\"#{schema_name}\".#{table_name}"
|
||||
sql = %{
|
||||
SELECT TRUE as has_column FROM pg_catalog.pg_attribute a
|
||||
WHERE
|
||||
a.attname = '#{column_name}'
|
||||
AND a.attnum > 0
|
||||
AND NOT a.attisdropped
|
||||
AND a.attrelid = '#{qualified_table_name}'::regclass::oid
|
||||
LIMIT 1
|
||||
}
|
||||
result = user.in_database[sql].first
|
||||
result && result[:has_column]
|
||||
end
|
||||
|
||||
# From Table#get_the_geom_type!, adapted to unregistered tables
|
||||
# returns type to run Table#get_the_geom_type! afterwards again, which
|
||||
# saves the type in table metadata
|
||||
def fix_the_geom_type!(schema_name, table_name)
|
||||
qualified_table_name = "\"#{schema_name}\".#{table_name}"
|
||||
|
||||
type = nil
|
||||
the_geom_data = user.in_database[%Q{
|
||||
SELECT a.attname, t.typname
|
||||
FROM pg_attribute a, pg_type t
|
||||
WHERE attrelid = '#{qualified_table_name}'::regclass
|
||||
AND attname = '#{THE_GEOM}'
|
||||
AND a.atttypid = t.oid
|
||||
AND a.attstattarget < 0
|
||||
LIMIT 1
|
||||
}].first
|
||||
return nil unless the_geom_data
|
||||
|
||||
if the_geom_data[:typname] != 'geometry'
|
||||
user.in_database.execute %{
|
||||
ALTER TABLE #{qualified_table_name} RENAME COLUMN "#{THE_GEOM}" TO "the_geom_str"
|
||||
}
|
||||
return nil
|
||||
end
|
||||
|
||||
geom_type = user.in_database[%Q{
|
||||
SELECT GeometryType(#{THE_GEOM})
|
||||
FROM #{qualified_table_name}
|
||||
WHERE #{THE_GEOM} IS NOT null
|
||||
LIMIT 1
|
||||
}].first
|
||||
|
||||
type = geom_type[:geometrytype].to_s.downcase if geom_type
|
||||
|
||||
# if the geometry is MULTIPOINT we convert it to POINT
|
||||
if type == 'multipoint'
|
||||
user.db_service.in_database_direct_connection(statement_timeout: STATEMENT_TIMEOUT) do |user_database|
|
||||
user_database.run("SELECT public.AddGeometryColumn('#{schema_name}', '#{table_name}','the_geom_simple',4326, 'GEOMETRY', 2);")
|
||||
user_database.run(%Q{UPDATE #{qualified_table_name} SET the_geom_simple = ST_GeometryN(the_geom,1);})
|
||||
user_database.run("SELECT DropGeometryColumn('#{schema_name}', '#{table_name}','the_geom');")
|
||||
user_database.run(%Q{ALTER TABLE #{qualified_table_name} RENAME COLUMN the_geom_simple TO the_geom;})
|
||||
end
|
||||
type = 'point'
|
||||
end
|
||||
|
||||
# if the geometry is LINESTRING or POLYGON we convert it to MULTILINESTRING or MULTIPOLYGON
|
||||
if %w(linestring polygon).include?(type)
|
||||
user.db_service.in_database_direct_connection(statement_timeout: STATEMENT_TIMEOUT) do |user_database|
|
||||
user_database.run("SELECT public.AddGeometryColumn('#{schema_name}', '#{table_name}','the_geom_simple',4326, 'GEOMETRY', 2);")
|
||||
user_database.run(%Q{UPDATE #{qualified_table_name} SET the_geom_simple = ST_Multi(the_geom);})
|
||||
user_database.run("SELECT DropGeometryColumn('#{schema_name}', '#{table_name}','the_geom');")
|
||||
user_database.run(%Q{ALTER TABLE #{qualified_table_name} RENAME COLUMN the_geom_simple TO the_geom;})
|
||||
|
||||
type = user_database[%Q{
|
||||
SELECT GeometryType(#{THE_GEOM})
|
||||
FROM #{qualified_table_name}
|
||||
WHERE #{THE_GEOM} IS NOT null
|
||||
LIMIT 1
|
||||
}].first[:geometrytype]
|
||||
end
|
||||
end
|
||||
|
||||
type
|
||||
end
|
||||
|
||||
# From Table#import_cleanup, with column schema checks adapted to unregistered tables
|
||||
def import_cleanup(schema_name, table_name)
|
||||
qualified_table_name = "\"#{schema_name}\".#{table_name}"
|
||||
|
||||
user.db_service.in_database_direct_connection(statement_timeout: STATEMENT_TIMEOUT) do |user_database|
|
||||
|
||||
# For consistency with regular imports, also eases testing
|
||||
Table.sanitize_columns(table_name, {database_schema: schema_name, connection: user_database})
|
||||
|
||||
# When tables are created using ogr2ogr they are added a ogc_fid or gid primary key
|
||||
# In that case:
|
||||
# - If cartodb_id already exists, remove ogc_fid
|
||||
# - If cartodb_id does not exist, treat this field as the auxiliary column
|
||||
aux_cartodb_id_column = [:ogc_fid, :gid].find do |col|
|
||||
valid_cartodb_id_candidate?(user, table_name, qualified_table_name, col)
|
||||
end
|
||||
|
||||
# Remove primary key
|
||||
existing_pk = user_database[%Q{
|
||||
SELECT c.conname AS pk_name
|
||||
FROM pg_class r, pg_constraint c, pg_namespace n
|
||||
WHERE r.oid = c.conrelid AND contype='p' AND relname = '#{table_name}'
|
||||
AND r.relnamespace = n.oid and n.nspname= '#{schema_name}'
|
||||
}].first
|
||||
existing_pk = existing_pk[:pk_name] unless existing_pk.nil?
|
||||
user_database.run(%Q{
|
||||
ALTER TABLE #{qualified_table_name} DROP CONSTRAINT "#{existing_pk}"
|
||||
}) unless existing_pk.nil?
|
||||
|
||||
# All normal fields casted to text
|
||||
varchar_columns = user_database[%Q{
|
||||
SELECT a.attname, t.typname
|
||||
FROM pg_attribute a, pg_type t
|
||||
WHERE attrelid = '#{qualified_table_name}'::regclass
|
||||
AND typname = 'varchar'
|
||||
AND a.atttypid = t.oid
|
||||
}].all
|
||||
|
||||
varchar_columns.each do |column|
|
||||
user_database.run(%Q{ALTER TABLE #{qualified_table_name} ALTER COLUMN "#{column[:attname]}" TYPE text})
|
||||
end
|
||||
|
||||
# If there's an auxiliary column, copy to cartodb_id and restart the sequence to the max(cartodb_id)+1
|
||||
if aux_cartodb_id_column.present?
|
||||
begin
|
||||
already_had_cartodb_id = false
|
||||
user_database.run(%Q{ALTER TABLE #{qualified_table_name} ADD COLUMN cartodb_id SERIAL})
|
||||
rescue
|
||||
already_had_cartodb_id = true
|
||||
end
|
||||
unless already_had_cartodb_id
|
||||
user_database.run(%Q{UPDATE #{qualified_table_name} SET cartodb_id = CAST(#{aux_cartodb_id_column} AS INTEGER)})
|
||||
cartodb_id_sequence_name = user_database["SELECT pg_get_serial_sequence('#{schema_name}.#{table_name}', 'cartodb_id')"].first[:pg_get_serial_sequence]
|
||||
max_cartodb_id = user_database[%Q{SELECT max(cartodb_id) FROM #{qualified_table_name}}].first[:max]
|
||||
# only reset the sequence on real imports.
|
||||
|
||||
if max_cartodb_id
|
||||
user_database.run("ALTER SEQUENCE #{cartodb_id_sequence_name} RESTART WITH #{max_cartodb_id + 1}")
|
||||
end
|
||||
end
|
||||
user_database.run(%Q{ALTER TABLE #{qualified_table_name} DROP COLUMN #{aux_cartodb_id_column}})
|
||||
end
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
def success?
|
||||
(!@failed && runner.success?)
|
||||
end
|
||||
|
||||
def etag
|
||||
runner.etag
|
||||
end
|
||||
|
||||
def last_modified
|
||||
runner.last_modified
|
||||
end
|
||||
|
||||
def checksum
|
||||
runner.checksum
|
||||
end
|
||||
|
||||
def move_to_schema(result, schema=DESTINATION_SCHEMA)
|
||||
# The new table to sync is moved to user schema to allow CartoDBfication.
|
||||
# This temporary table should not be registered (check ghost_tables_manager.rb)
|
||||
return self if schema == result.schema
|
||||
database.execute(%Q{
|
||||
ALTER TABLE "#{result.schema}"."#{result.table_name}"
|
||||
SET SCHEMA "#{user.database_schema}"
|
||||
})
|
||||
end
|
||||
|
||||
def rename(current_name, new_name)
|
||||
database.execute(%Q{
|
||||
ALTER TABLE "#{user.database_schema}"."#{current_name}"
|
||||
RENAME TO #{new_name}
|
||||
})
|
||||
end
|
||||
|
||||
def drop(table_name)
|
||||
database.execute(%Q(DROP TABLE "#{user.database_schema}"."#{table_name}"))
|
||||
end
|
||||
|
||||
def exists?(table_name)
|
||||
database.table_exists?(table_name)
|
||||
end
|
||||
|
||||
def results
|
||||
runner.results
|
||||
end
|
||||
|
||||
def error_code
|
||||
@error_code || runner.results.map(&:error_code).compact.first
|
||||
end
|
||||
|
||||
def runner_log_trace
|
||||
runner.results.map(&:log_trace).compact.first
|
||||
end
|
||||
|
||||
def error_message
|
||||
''
|
||||
end
|
||||
|
||||
def temporary_name_for(table_name)
|
||||
"#{table_name}_to_be_deleted"
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def valid_cartodb_id_candidate?(user, table_name, qualified_table_name, col_name)
|
||||
return false unless column_names(user, table_name).include?(col_name)
|
||||
user.transaction_with_timeout(statement_timeout: STATEMENT_TIMEOUT, as: :superuser) do |db|
|
||||
return db["SELECT 1 FROM #{qualified_table_name} WHERE #{col_name} IS NULL LIMIT 1"].first.nil?
|
||||
end
|
||||
end
|
||||
|
||||
def column_names(user, table_name)
|
||||
user.in_database.schema(table_name, schema: user.database_schema).map { |row| row[0] }
|
||||
rescue => e
|
||||
CartoDB::Logger.error(
|
||||
message: 'Error in column_names from sync adapter',
|
||||
exception: e,
|
||||
user: user,
|
||||
table: table_name
|
||||
)
|
||||
[]
|
||||
end
|
||||
|
||||
attr_reader :table_name, :runner, :database, :user
|
||||
end
|
||||
end
|
||||
end
|
||||
Reference in New Issue
Block a user