AMBARI-4436. ambari-server upgrade takes a long time if there are many command entries in the DB. (mpapirkovskyy)

ambari-server/src/main/resources/upgrade/ddl/Ambari-DDL-Oracle-UPGRADE.sql

@@ -52,294 +52,11 @@ ALTER TABLE confgroupclusterconfigmapping ADD CONSTRAINT FK_confgroupclusterconf
 ALTER TABLE configgrouphostmapping ADD CONSTRAINT FK_configgrouphostmapping_configgroup_id FOREIGN KEY (config_group_id) REFERENCES configgroup (group_id);
 ALTER TABLE configgrouphostmapping ADD CONSTRAINT FK_configgrouphostmapping_host_name FOREIGN KEY (host_name) REFERENCES hosts (host_name);
 
+-- Don't set not null constraint
+-- ALTER TABLE stage MODIFY (cluster_host_info NOT NULL);
 
-UPDATE
-  stage sd
-SET
-  (sd.cluster_host_info) =
-  (
-    SELECT DISTINCT
-      (dbms_lob.substr(ec.command, dbms_lob.instr(ec.command,
-      '636f6e66696775726174696f6e73')   - dbms_lob.instr(ec.command,
-      '636c7573746572486f7374496e666f') - 19, dbms_lob.instr(ec.command,
-      '636c7573746572486f7374496e666f') + 17) )
-    FROM
-      execution_command ec ,
-      host_role_command hrc,
-      stage ss
-    WHERE
-      ec.task_id       = hrc.task_id
-    AND hrc.stage_id   = ss.stage_id
-    AND hrc.request_id = ss.request_id
-    AND ss.stage_id    = sd.stage_id
-    AND ss.request_id  = sd.request_id
-  );
-  
-ALTER TABLE stage MODIFY (cluster_host_info NOT NULL);
-
-
-
-
-CREATE OR REPLACE PACKAGE compress_cluster_host_info_pkg AS
-  TYPE nested_vachar2_tbl_type IS TABLE OF VARCHAR2(4000);
-  FUNCTION compress_cluster_host_info(p_cluster_host_info BLOB) RETURN BLOB;
-  FUNCTION get_keys(p_cluster_host_info VARCHAR2) RETURN VARCHAR2TBL PIPELINED;
-  FUNCTION get_value(p_cluster_host_info BLOB, p_param_key VARCHAR2) RETURN VARCHAR2;
-  FUNCTION split_to_array(p_string VARCHAR2,p_sep VARCHAR2) RETURN nested_vachar2_tbl_type;
-  FUNCTION index_of(p_arr  nested_vachar2_tbl_type, p_item VARCHAR2) RETURN INTEGER;
-  FUNCTION to_indexed_array(p_arr nested_vachar2_tbl_type, p_dict_arr nested_vachar2_tbl_type) RETURN nested_vachar2_tbl_type;
-  FUNCTION To_mapped_indexed_array(p_arr nested_vachar2_tbl_type) RETURN nested_vachar2_tbl_type;
-  FUNCTION Nested_table_to_string(p_arr nested_vachar2_tbl_type, p_sep VARCHAR2) RETURN VARCHAR2;
-END compress_cluster_host_info_pkg;
-
-/
-
-CREATE OR REPLACE PACKAGE BODY compress_cluster_host_info_pkg AS
-  c_regex_pattern CONSTANT VARCHAR2(29) := '":\[[a-zA-Z0-9@:/":.,-]{1,}\]';
-  
-  PROCEDURE print_nested_table(p_arr NESTED_VACHAR2_TBL_TYPE)
-  AS
-  BEGIN
-      FOR i IN p_arr.FIRST..p_arr.LAST LOOP
-          dbms_output.put_line(i
-                               || ': '
-                               || P_arr(i));
-      END LOOP;
-  END print_nested_table;
-  FUNCTION Get_keys(p_cluster_host_info VARCHAR2)
-  RETURN VARCHAR2TBL PIPELINED
-  AS
-  BEGIN
-      FOR r IN (SELECT param_key
-                FROM   (SELECT substr(regexp_substr(regexp_replace(
-                                                    p_cluster_host_info
-                                                    , c_regex_pattern
-                                                            , ' '),
-                                                      '[^ ]+'
-                                      , 1, LEVEL), 3) AS param_key
-                        FROM dual
-                        CONNECT BY LEVEL <= regexp_count(regexp_replace(
-                                                         p_cluster_host_info,
-                                                         c_regex_pattern, ' '),
-                                            '[^ ]+'))
-                WHERE  param_key IS NOT NULL
-                       AND NOT param_key LIKE '%ambari_db_rca%') LOOP
-          PIPE ROW(r.param_key);
-      END LOOP;
-  END get_keys;
-  FUNCTION get_value(p_cluster_host_info BLOB,
-                     p_param_key         VARCHAR2)
-  RETURN VARCHAR2
-  AS
-    v_param_value VARCHAR2(32767);
-  BEGIN
-      SELECT regexp_substr(utl_raw.Cast_to_varchar2(p_cluster_host_info), '"'
-                                                                          ||
-                   p_param_key
-                                                                          ||
-             '":\[["a-z0-9., ]{1,}]')
-      INTO   v_param_value
-      FROM   dual;
-
-      SELECT substr(v_param_value, length(p_param_key) + 5,
-                    dbms_lob.Getlength(v_param_value) - length(p_param_key) - 5)
-      INTO   v_param_value
-      FROM   dual;
-
-      RETURN v_param_value;
-  END get_value;
-  
-  FUNCTION compress_cluster_host_info (p_cluster_host_info BLOB)
-  RETURN BLOB
-  AS
-    CURSOR cur1(
-      p_param_name VARCHAR2) IS
-      SELECT *
-      FROM   (SELECT column_value
-                            AS
-                            param_name,
-  compress_cluster_host_info_pkg.get_value(p_cluster_host_info, column_value) AS
-  param_value
-  FROM   TABLE(compress_cluster_host_info_pkg.get_keys((
-                      utl_raw.cast_to_varchar2(p_cluster_host_info) )))) a
-  WHERE  ( a.param_name = p_param_name
-          OR ( p_param_name IS NULL
-               AND a.param_name NOT IN ( 'all_hosts', 'all_ping_ports' ) ) );
-  l_result                BLOB;
-  l_raw                   RAW(32767);
-  l_all_hosts             NESTED_VACHAR2_TBL_TYPE;
-  l_all_ping_ports        NESTED_VACHAR2_TBL_TYPE;
-  l_compressed_ping_ports NESTED_VACHAR2_TBL_TYPE;
-  l_indexed               NESTED_VACHAR2_TBL_TYPE;
-  BEGIN
-    dbms_lob.createtemporary(l_result, FALSE);
-
-    dbms_lob.OPEN(l_result, dbms_lob.lob_readwrite);
-
-    FOR r IN cur1('all_hosts') LOOP
-      l_all_hosts := split_to_array(r.param_value, ',');
-    END LOOP;
-
-    FOR r IN cur1('all_ping_ports') LOOP
-      l_all_ping_ports := split_to_array(r.param_value, ',');
-
-      dbms_output.put_line(r.param_value);
-    END LOOP;
-
-    l_compressed_ping_ports := to_mapped_indexed_array(l_all_ping_ports);
-
-    l_raw := utl_raw.cast_to_raw('{');
-
-    dbms_lob.writeappend(l_result, utl_raw.length(l_raw), l_raw);
-
-    FOR r IN cur1(NULL) LOOP
-      dbms_output.put_line(r.param_name);
-
-      l_indexed := to_indexed_array(split_to_array(r.param_value, ','),
-                 l_all_hosts);
-
-      l_raw := utl_raw.cast_to_raw('"'
-                                 || r.param_name
-                                 || '":["'
-                                 || nested_table_to_string(l_indexed, ',')
-                                 || '"],');
-
-      dbms_lob.writeappend(l_result, utl_raw.Length(l_raw), l_raw);
-    END LOOP;
-
-    l_raw := utl_raw.cast_to_raw('"all_hosts":['
-                             || nested_table_to_string(l_all_hosts, ',')
-                             || '],');
-
-    dbms_lob.writeappend(l_result, utl_raw.length(l_raw), l_raw);
-
-    l_raw := utl_raw.Cast_to_raw('"all_ping_ports":['
-                             || Nested_table_to_string(
-                                l_compressed_ping_ports,
-                                ',')
-                             || ']');
-
-    dbms_lob.Writeappend(l_result, utl_raw.Length(l_raw), l_raw);
-
-    l_raw := utl_raw.Cast_to_raw('}');
-
-    dbms_lob.writeappend(l_result, utl_raw.Length(l_raw), l_raw);
-
-    dbms_lob.CLOSE(l_result);
-
-    RETURN l_result;
-    
-  END compress_cluster_host_info;
-
-  FUNCTION split_to_array(p_string VARCHAR2,
-                          p_sep    VARCHAR2)
-  RETURN NESTED_VACHAR2_TBL_TYPE
-  AS
-    l_result NESTED_VACHAR2_TBL_TYPE;
-  BEGIN
-      SELECT Regexp_substr(p_string, '[^,]+', 1, LEVEL)
-      BULK   COLLECT INTO l_result
-      FROM   dual
-      CONNECT BY Instr(p_string, p_sep, 1, LEVEL - 1) > 0;
-
-      RETURN l_result;
-  END split_to_array;
-  
-  FUNCTION index_of(p_arr  NESTED_VACHAR2_TBL_TYPE,
-                    p_item VARCHAR2)
-  RETURN INTEGER
-  AS
-  BEGIN
-      FOR i IN p_arr.FIRST..p_arr.LAST LOOP
-          IF p_arr(i) = p_item THEN
-            RETURN i - 1;
-          END IF;
-      END LOOP;
-  END index_of;
-  
-  FUNCTION to_indexed_array(p_arr      NESTED_VACHAR2_TBL_TYPE,
-                            p_dict_arr NESTED_VACHAR2_TBL_TYPE)
-  RETURN NESTED_VACHAR2_TBL_TYPE
-  AS
-    l_index  INTEGER;
-    l_result NESTED_VACHAR2_TBL_TYPE := Nested_vachar2_tbl_type();
-  BEGIN
-      FOR i IN p_arr.first..p_arr.last LOOP
-          l_index := Index_of(p_dict_arr, P_arr(i));
-
-          l_result.Extend(1);
-
-          L_result(i) := l_index;
-      END LOOP;
-
-      RETURN l_result;
-  END to_indexed_array;
-  FUNCTION to_mapped_indexed_array(p_arr NESTED_VACHAR2_TBL_TYPE)
-  RETURN NESTED_VACHAR2_TBL_TYPE
-  AS
-    v_result       NESTED_VACHAR2_TBL_TYPE := Nested_vachar2_tbl_type();
-    v_curr_indexes VARCHAR2(32767);
-    v_prev_val     VARCHAR2(32767);
-  BEGIN
-      FOR i IN p_arr.first..p_arr.last LOOP
-          IF P_arr(i) <> v_prev_val THEN
-            v_result.extend(1);
-
-            V_result(v_result.last) := '"'
-                                       || v_curr_indexes
-                                       || '"';
-
-            v_curr_indexes := NULL;
-          END IF;
-
-          IF v_curr_indexes IS NULL THEN
-            v_curr_indexes := substr(p_arr(i), 2, length(p_arr(i)) - 2)
-                              || ':'
-                              || to_char(i - 1);
-          ELSE
-            v_curr_indexes := v_curr_indexes
-                              || ','
-                              || to_char(i - 1);
-          END IF;
-
-          v_prev_val := p_arr(i);
-      END LOOP;
-
-      IF v_curr_indexes IS NOT NULL THEN
-        v_result.extend(1);
-
-        V_result(v_result.LAST) := '"' || v_curr_indexes || '"';
-      END IF;
-
-      RETURN v_result;
-  END to_mapped_indexed_array;
-  FUNCTION nested_table_to_string(p_arr NESTED_VACHAR2_TBL_TYPE,
-                                  p_sep VARCHAR2)
-  RETURN VARCHAR2
-  AS
-    v_result VARCHAR2(32767);
-  BEGIN
-      v_result := p_arr(1);
-
-      FOR i IN p_arr.FIRST + 1 ..p_arr.LAST LOOP
-          v_result := v_result
-                      || ','
-                      || p_arr(i);
-      END LOOP;
-
-      RETURN v_result;
-  END nested_table_to_string;
-END compress_cluster_host_info_pkg; 
-/
-
---Compress cluster host info
-UPDATE stage s
-SET s.cluster_host_info = compress_cluster_host_info_pkg.compress_cluster_host_info(s.cluster_host_info)
-WHERE dbms_lob.instr(cluster_host_info, utl_raw.cast_to_raw('ambari_db_rca'), 1, 1) > 0;
-
---Drop compression package
-DROP PACKAGE compress_cluster_host_info_pkg;
-
+-- Abort all tasks in progress due to format change
+UPDATE host_role_command SET status = 'ABORTED' WHERE status IN ('PENDING', 'QUEUED', 'IN_PROGRESS');
 
 ALTER TABLE hosts DROP COLUMN disks_info;
 

ambari-server/src/main/resources/upgrade/ddl/Ambari-DDL-Postgres-UPGRADE-1.3.0.sql

@@ -136,200 +136,10 @@ ALTER TABLE ambari.configgrouphostmapping ADD CONSTRAINT FK_configgrouphostmappi
 ALTER TABLE ambari.hostcomponentdesiredstate ADD COLUMN admin_state VARCHAR(32);
 ALTER TABLE ambari.hostcomponentdesiredstate ADD COLUMN passive_state VARCHAR(32) NOT NULL DEFAULT 'ACTIVE'
 
---Move cluster host info for old execution commands to stage table
-UPDATE ambari.stage sd
-  SET 
-    cluster_host_info = substring(ec.command, position('clusterHostInfo' in ec.command) + 17, position('configurations' in ec.command) - position('clusterHostInfo' in ec.command) - 19)
-  FROM
-    ambari.execution_command ec,
-    ambari.host_role_command hrc,
-    ambari.stage ss
-  WHERE ec.task_id = hrc.task_id
-  AND hrc.stage_id = ss.stage_id
-  AND hrc.request_id = ss.request_id
-  AND sd.cluster_host_info IS NULL;
-  
-
---Set cluster_host_info column mandatory
-ALTER TABLE ambari.stage ALTER COLUMN cluster_host_info SET NOT NULL;
-
---Compress cluster host info-----------------------------
-CREATE OR REPLACE FUNCTION get_keys(p_cluster_host_info text)
-  RETURNS setof text AS
-$_$
-DECLARE
-v_r text;
-BEGIN
-  FOR v_r IN (SELECT substr(key_tokens,3,length(key_tokens)) AS cluster_host_info_key
-  FROM regexp_split_to_table(p_cluster_host_info, E'":\[[a-z0-9":.,-]{1,}\]') AS key_tokens
-   WHERE key_tokens NOT LIKE '%ambari_db_rca_%') LOOP
-     RETURN NEXT v_r;
-  END LOOP;
-END;
-$_$ LANGUAGE plpgsql;
-
-
-
-CREATE OR REPLACE FUNCTION get_value(p_cluster_host_info text, p_param_key text)
-  RETURNS text AS
-$_$
-
-DECLARE
-v_param_value text;
-BEGIN
-
-	SELECT regexp_matches(p_cluster_host_info,
-
-	 '"' || p_param_key || E'":\[["a-z0-9., ]{1,}]') into v_param_value;
-
-	SELECT substring(v_param_value, length(p_param_key) + 9, length(v_param_value) - length(p_param_key) - 11) into v_param_value;
-
-	RETURN v_param_value;
-	
-END;
-$_$ LANGUAGE plpgsql;
-
-CREATE OR REPLACE FUNCTION compress_cluster_host_info(p_stage_id ambari.stage.stage_id%type, p_request_id ambari.stage.request_id%type) RETURNS text AS
-$_$
-DECLARE
-
-cur1 CURSOR(p_param_name text) IS 
-
-  select a.param_key, string_to_array(get_value(cast(cluster_host_info as text), a.param_key), ',') as param_value
-  from (
-    select s.stage_id, request_id, get_keys(cast(cluster_host_info as text)) as param_key, s.cluster_host_info from ambari.stage s) a
-      where stage_id = p_stage_id
-      and request_id = p_request_id
-      and (a.param_key = p_param_name or (p_param_name is null and a.param_key not in ('all_hosts', 'all_ping_ports')));
-
-v_all_hosts text[];
-v_all_ping_ports text[];
-v_indexed integer[];
-v_r record;
-v_param_key text;
-v_compressed_ping_ports text[];
-v_compressed_cluster_host_info text;
-
-BEGIN
-
-  open cur1('all_hosts');
-  fetch cur1 into v_param_key, v_all_hosts;
-  close cur1;
-
-  open cur1('all_ping_ports');
-  fetch cur1 into v_param_key, v_all_ping_ports;
-  close cur1;
-
-  v_compressed_cluster_host_info = '{';
-
-  for v_r in cur1(null) loop
-    v_indexed = to_indexed_array(v_r.param_value, v_all_hosts);
-    select v_compressed_cluster_host_info || '"' || v_r.param_key || '":["' || array_to_string(v_indexed, ',') || '"],'
-    into v_compressed_cluster_host_info;
-
-  end loop;
-
-  v_compressed_ping_ports = to_mapped_indexed_array(v_all_ping_ports);
-
-  v_compressed_cluster_host_info = v_compressed_cluster_host_info || '"all_hosts":["' || array_to_string(v_all_hosts, ',') || '"],';
-
-  v_compressed_cluster_host_info = v_compressed_cluster_host_info || '"all_ping_ports":["' || array_to_string(v_compressed_ping_ports, ',') || '"]';
-
-  v_compressed_cluster_host_info = v_compressed_cluster_host_info || '}';
-
-  return v_compressed_cluster_host_info;
-
-  
-END;
-$_$ LANGUAGE plpgsql;
-
-CREATE OR REPLACE FUNCTION index_of(p_arr text[], p_item text)
-RETURNS INT AS 
-$_$
-DECLARE
-  v_index integer;
-BEGIN
-
-    SELECT i-1
-    into v_index
-      FROM generate_subscripts(p_arr, 1) AS i
-     WHERE p_arr[i] = p_item
-  ORDER BY i;
-
-  RETURN v_index;
-END;
-$_$ LANGUAGE plpgsql;
-
-CREATE OR REPLACE FUNCTION to_indexed_array(arr text[], dict_array text[])
-RETURNS integer[] AS 
-$_$
-
-DECLARE
-
-v_result integer[];
-v_index_of integer;
-
-BEGIN
-
-  FOR i IN array_lower(arr, 1)..array_upper(arr, 1)
-    LOOP
-        v_index_of = index_of(dict_array, arr[i]);
-        select array_append(v_result, v_index_of) into v_result;
-    END LOOP;
-
-  RETURN v_result; 
-    
-END;
-$_$ LANGUAGE plpgsql;
-
-CREATE OR REPLACE FUNCTION to_mapped_indexed_array(p_arr text[])
-RETURNS text[] AS 
-$_$
-DECLARE
-v_result text[];
-v_r record;
-v_curr_indexes text;
-v_prev_val text;
-BEGIN
-
-  FOR v_r in (select (row_number() OVER (ORDER BY 1)) -1 AS ind, x AS val from (select unnest(p_arr) AS x) a) LOOP
-
-    if v_r.val <> v_prev_val then
-      v_result = array_append(v_result, v_curr_indexes);
-      v_curr_indexes = null;
-    end if;
-
-    if v_curr_indexes is null then
-      v_curr_indexes = v_r.val || ':' || v_r.ind;
-    else
-      v_curr_indexes = v_curr_indexes || ',' || v_r.ind;
-    end if;
-
-    v_prev_val = v_r.val;
-    
-  END LOOP;
-
-  if v_curr_indexes is not null then
-    v_result = array_append(v_result, v_curr_indexes);
-  end if;
-  
-  RETURN v_result; 
-    
-END;
-$_$ LANGUAGE plpgsql;
-
---Update cluster host info to compressed values
-UPDATE ambari.stage s
-SET cluster_host_info = (decode(replace(compress_cluster_host_info(stage_id, request_id), E'\\', E'\\\\'), 'escape'))
-WHERE s.cluster_host_info LIKE '%ambari_db_rca%';
-
---Drop compression functions
-DROP FUNCTION get_keys;
-DROP FUNCTION get_value;
-DROP FUNCTION compress_cluster_host_info;
-DROP FUNCTION to_indexed_array;
-DROP FUNCTION to_mapped_indexed_array;
+-- Set cluster_host_info column mandatory -- disabled due to update logic change
+-- ALTER TABLE ambari.stage ALTER COLUMN cluster_host_info SET NOT NULL;
 
+UPDATE ambari.host_role_command SET status = 'ABORTED' WHERE status IN ('PENDING', 'QUEUED', 'IN_PROGRESS');
 
 ALTER TABLE ambari.hosts DROP COLUMN disks_info;