Revert "fix"

This reverts commit c3322548.

Pyspark_job/sqoop_export/dwt_amazon_report_pg.py

@@ -23,35 +23,25 @@ if __name__ == '__main__':
     engine = DBUtil.get_db_engine(db_type, site_name)
 
     with engine.connect() as connection:
-        # 不再用 like {master_tb} including all 复制结构：显式建表，新字段直接写进结构里，
-        # 不需要对线上分布式表 master_tb 做 ALTER TABLE ADD COLUMN（容易触发 Citus 分布式死锁）；
-        # 新字段随着导出完成后的换表流程一起生效。建表时不建索引（导出完成后由 exchange_tb
-        # 从 master_tb 复制现有索引重建，整表建索引比逐行插入维护索引快很多）
         sql = f"""
-            DROP TABLE IF EXISTS "public"."{export_tb}";
-            CREATE TABLE "public"."{export_tb}" (
-              "asin" varchar(20) COLLATE "pg_catalog"."default" NOT NULL,
-              "monthly_sales" text COLLATE "pg_catalog"."default" NOT NULL,
-              "zr_count" text COLLATE "pg_catalog"."default" NOT NULL,
-              "sp_count" text COLLATE "pg_catalog"."default" NOT NULL,
-              "total_count" text COLLATE "pg_catalog"."default" NOT NULL,
-              "date_info_list" text COLLATE "pg_catalog"."default" NOT NULL,
-              "created_at" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-              "updated_at" timestamp(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
-              "weekly_sales" text COLLATE "pg_catalog"."default",
-              "weekly_views" text COLLATE "pg_catalog"."default",
-              "monthly_views" text COLLATE "pg_catalog"."default",
-              "variation_monthly_sales" text COLLATE "pg_catalog"."default" NOT NULL
+            drop table if exists {export_tb};
+            create table if not exists {export_tb} 
+            (
+            like {master_tb} including ALL       
             );
-            COMMENT ON COLUMN "public"."{export_tb}"."asin" IS 'asin';
-            COMMENT ON COLUMN "public"."{export_tb}"."monthly_sales" IS 'Amazon月销';
-            COMMENT ON COLUMN "public"."{export_tb}"."zr_count" IS 'zr词数';
-            COMMENT ON COLUMN "public"."{export_tb}"."sp_count" IS 'sp词数';
-            COMMENT ON COLUMN "public"."{export_tb}"."total_count" IS '总词数';
-            COMMENT ON COLUMN "public"."{export_tb}"."date_info_list" IS 'date_info';
-            COMMENT ON COLUMN "public"."{export_tb}"."variation_monthly_sales" IS '变体销量月度历史(逗号分隔,对齐date_info_list)';
 
             SELECT create_distributed_table('{export_tb}', 'asin');
+
+            DO $$
+            DECLARE
+                index_record RECORD;
+            BEGIN
+                FOR index_record IN (SELECT indexname FROM pg_indexes WHERE tablename = '{export_tb}') 
+                LOOP
+                    EXECUTE 'DROP INDEX IF EXISTS ' || index_record.indexname;
+                END LOOP;
+            END;
+            $$;
         """
         connection.execute(sql)