code: def upgrade():
      with op.batch_alter_table("predictor", schema=None) as batch_op:
          batch_op.add_column(
              sa.Column("integration_id", sa.Integer(), nullable=True)
          )
          batch_op.add_column(
              sa.Column("fetch_data_query", sa.String(), nullable=True)
          )
          batch_op.create_foreign_key(
              "fk_integration_id",
              "integration",
              ["integration_id"],
              ["id"],
          )
  
      conn = op.get_bind()
  
      conn.execute(
          text(
              """
          insert into integration (name, data, company_id, created_at, updated_at)
              select
                  'files' as name,
                  '{}' as data,
                  company_id,
                  '2022-05-01 00:00:00.000000' as created_at,
                  '2022-05-01 00:00:00.000000' as updated_at
              from (select distinct company_id from integration) t1
      """
          )
      )
  
      predictors = conn.execute(
          text(
              """
          select t1.id, t1.company_id, t2.data
          from predictor t1 left join dataset t2 on t1.id = t2.id
          where dataset_id is not null and t2.data is not null
      """
          )
      ).fetchall()
      for row in predictors:
          data = row["data"]
          try:
              data = json.loads(data)
          except Exception:
              continue
  
          if "source_type" not in data:
              continue
          integration_name = data.get("source_type")
          if (
              isinstance(integration_name, str) is False
              or len(integration_name) == 0
          ):
              continue
          if integration_name.lower() == "file":
              integration_name = "files"
  
          fetch_data_query = data.get("source")
          if isinstance(fetch_data_query, dict) is False:
              continue
  
          if integration_name == "files":
              file_name = fetch_data_query.get("mindsdb_file_name")
              if (
                  isinstance(file_name, str) is False
                  or len(file_name) == 0
              ):
                  continue
              fetch_data_query = f"select * from {file_name}"
          else:
              fetch_data_query = fetch_data_query.get("query")
              if (
                  isinstance(fetch_data_query, str) is False
                  or len(fetch_data_query) == 0
              ):
                  continue
  
          query = """
              select id
              from integration
              where company_id = :company_id and lower(name) = lower(:name)
          """
          if row["company_id"] is None:
              query = """
                  select id
                  from integration
                  where company_id is null and lower(name) = lower(:name)
              """
          integration = conn.execute(
              text(query),
              {
                  "company_id": row["company_id"],
                  "name": integration_name,
              },
          ).fetchone()
          if integration is None:
              continue
  
          conn.execute(
              text(
                  """
              update predictor
              set integration_id = :integration_id, fetch_data_query = :fetch_data_query
              where id = :predictor_id
          """
              ),
              {
                  "integration_id": integration.id,
                  "fetch_data_query": fetch_data_query,
                  "predictor_id": row["id"],
              },
          )
  
      with op.batch_alter_table("predictor", schema=None) as batch_op:
          batch_op.drop_column("dataset_id")
      with op.batch_alter_table("file", schema=None) as batch_op:
          batch_op.drop_constraint(
              "unique_file_name_company_id", type_="unique"
          )
          batch_op.drop_constraint("fk_analysis_id", type_="foreignkey")
          batch_op.drop_column("analysis_id")
      with op.batch_alter_table("dataset", schema=None) as batch_op:
          batch_op.drop_constraint(
              "fk_ds_analysis_id", type_="foreignkey"
          )
          batch_op.drop_column("analysis_id")
      op.drop_table("analysis")
      op.drop_table("dataset")
  
imports: import json
  from alembic import op
  import sqlalchemy
  from sqlalchemy.sql import text
label_desc: Use image node_id 1 for calling a global function with example usage: upgrade() without return types