"""
Parquet 课程 · 文件解剖器（第 2–5 课共用）

把一个 Parquet 文件的内核打印出来：footer 概况、每列体积与编码、row group 划分、
每个 row group 的 min/max 统计、Page Index 与 Bloom filter 是否存在。

    python inspect_parquet.py data/iot_sorted.parquet

看不懂输出的哪一段，就回到对应课次：结构看第 2 课，编码看第 3 课，统计看第 4 课。
"""
import sys, os
import duckdb
import pyarrow.parquet as pq

path = sys.argv[1] if len(sys.argv) > 1 else "data/iot_sorted.parquet"
con = duckdb.connect()
md = pq.ParquetFile(path).metadata

print(f"文件      : {path}  ({os.path.getsize(path)/1024/1024:.2f} MB)")
print(f"行数      : {md.num_rows:,}")
print(f"row group : {md.num_row_groups} 个")
print(f"footer    : {md.serialized_size:,} 字节  ({md.serialized_size/os.path.getsize(path)*100:.3f}% 的文件)")
print(f"写入者    : {md.created_by}\n")

print("== 每列体积与编码 ==")
print(con.sql(f"""
  SELECT path_in_schema AS col, any_value(type) AS phys_type,
         round(sum(total_compressed_size)/1024.0/1024, 2) AS comp_mb,
         round(sum(total_uncompressed_size)/1024.0/1024, 2) AS raw_mb,
         round(sum(total_uncompressed_size)*1.0/sum(total_compressed_size), 1) AS x,
         any_value(encodings) AS encodings,
         any_value(compression) AS codec
  FROM parquet_metadata('{path}') GROUP BY 1 ORDER BY comp_mb DESC"""))

print("== row group 划分 ==")
print(con.sql(f"""
  SELECT row_group_id AS rg, any_value(row_group_num_rows) AS rows,
         round(any_value(row_group_bytes)/1024.0/1024, 2) AS rg_mb
  FROM parquet_metadata('{path}') GROUP BY 1 ORDER BY 1"""))

print("== 前 3 个 row group 的统计信息（决定能不能跳过）==")
print(con.sql(f"""
  SELECT row_group_id AS rg, path_in_schema AS col,
         stats_min_value AS min, stats_max_value AS max, stats_null_count AS nulls
  FROM parquet_metadata('{path}')
  WHERE row_group_id < 3 AND path_in_schema IN ('ts','device_id','value')
  ORDER BY rg, col"""))

print("== Page Index / Bloom filter ==")
cc0 = md.row_group(0).column(0)
print(f"Page Index (ColumnIndex/OffsetIndex): {cc0.has_column_index} / {cc0.has_offset_index}")
print(con.sql(f"""
  SELECT path_in_schema AS col,
         count(bloom_filter_offset) FILTER (bloom_filter_offset IS NOT NULL) AS bloom_rg_count
  FROM parquet_metadata('{path}') GROUP BY 1 ORDER BY 1"""))