feature-benchmark: add a pushdown benchmark

Author: DAlperin

misc/python/materialize/feature_benchmark/scenarios/benchmark_main.py

@@ -117,6 +117,52 @@ def benchmark(self) -> MeasurementSource:
 """)
 
 
+class ProjectionPushdown(Scenario):
+    """Hydrate an MV that reads one column from a 32-column persist table.
+
+    Exercises parquet ProjectionMask pushdown: persist's decoder skips the 31
+    columns the MV does not demand. Requires the
+    `storage_source_enable_column_projection` dyncfg, which `init()` enables.
+    """
+
+    SCALE = 5
+
+    def init(self) -> list[Action]:
+        return [
+            TdAction("""
+$ postgres-connect name=mz_system url=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr}
+$ postgres-execute connection=mz_system
+ALTER SYSTEM SET storage_source_enable_column_projection = true;
+
+> CREATE CLUSTER pp_cluster SIZE 'scale=1,workers=16', REPLICATION FACTOR 1
+"""),
+        ]
+
+    def benchmark(self) -> MeasurementSource:
+        n = self.n()
+        column_defs = ", ".join(f"f{i} INTEGER" for i in range(32))
+        select_cols = ", ".join(
+            f"((generate_series * {17 + i * 7}) % 1000003)::int AS f{i}"
+            for i in range(32)
+        )
+        return Td(f"""
+> DROP TABLE IF EXISTS pp_wide CASCADE
+> CREATE TABLE pp_wide ({column_defs})
+> INSERT INTO pp_wide SELECT {select_cols} FROM generate_series(1, {n})
+> ALTER CLUSTER pp_cluster SET (REPLICATION FACTOR 0)
+> CREATE MATERIALIZED VIEW pp_narrow IN CLUSTER pp_cluster AS SELECT f0 FROM pp_wide
+> SET CLUSTER = pp_cluster
+> SELECT 1
+  /* A */
+1
+> ALTER CLUSTER pp_cluster SET (REPLICATION FACTOR 1)
+> SELECT COUNT(*) FROM pp_narrow
+  /* B */
+{n}
+> SET CLUSTER = default
+""")
+
+
 class FastPathFilterIndex(FastPath):
     """Measure the time it takes for the fast path to filter our all rows from a materialized view using an index and return"""