Untitled

full_tbl = "iceberg.sompodatabase2.T001PLPYIN"
date_col = "CONFIRM_DATE"
pk_cols = ["FIRM_CODE", "COMPANY_CODE", "PRODUCT_NO", "POLICY_NO", "RENEWAL_NO", "ENDORS_NO"]

# --- 2️⃣ Partition'ı aylık olarak değiştir ---
spark.sql(f"""
  ALTER TABLE {full_tbl}
  ADD PARTITION FIELD months({date_col})
""")

print("[OK] Partition aylık olarak güncellendi.")

# --- 3️⃣ Bloom filter ayarlarını koru veya güncelle ---
bloom_cols = pk_cols + [date_col]
cols_str = ",".join(bloom_cols)
spark.sql(f"""
  ALTER TABLE {full_tbl}
  SET TBLPROPERTIES (
    'bloom_filter.columns' = '{cols_str}',
    'bloom_filter.fpp' = '0.02'
  )
""")

print("[OK] Bloom filter ayarları uygulandı.")

# --- 4️⃣ Dosyaları binpack ile optimize et ---
spark.sql(f"""
  CALL iceberg.system.rewrite_data_files(
    table => '{full_tbl}',
    strategy => 'binpack',
    options => map(
      'rewrite-all', 'true',
      'target-file-size-bytes', '268435456',
      'max-concurrent-file-group-rewrites', '16',
      'partial-progress.enabled', 'true',
      'partial-progress.max-commits', '20',
      'rewrite-job-order', 'files-desc'
    )
  )
""").show(truncate=False)

print("[OK] Data dosyaları yeniden yazıldı ve optimize edildi.")

# --- 5️⃣ Manifest dosyalarını da sıkıştır ---
spark.sql(f"""
  CALL iceberg.system.rewrite_manifests(
    table => '{full_tbl}'
  )
""")

print(f"[OK] Tüm tablo optimize edildi: {full_tbl}")

spark.stop()