feat(sql): support USING zonemap with distributed and consolidated build paths

docs/src/operations/ddl/create-index.md

@@ -22,10 +22,11 @@ The command uses the `ALTER TABLE` syntax to add an index.
 
 The following index methods are supported:
 
-| Method  | Description                                                                 |
-|---------|-----------------------------------------------------------------------------|
-| `btree` | B-tree index for efficient range queries and point lookups on scalar columns. |
-| `fts`   | Full-text search (inverted) index for text search on string columns.        |
+| Method    | Description                                                                 |
+|-----------|-----------------------------------------------------------------------------|
+| `btree`   | B-tree index for efficient range queries and point lookups on scalar columns. |
+| `fts`     | Full-text search (inverted) index for text search on string columns.        |
+| `zonemap` | Lightweight min/max statistics per zone for fragment pruning on a scalar column. |
 
 ## Options
 
@@ -42,6 +43,16 @@ For the `btree` method, the following options are supported:
 | `rows_per_range` | Long   | The number of rows per range when built using range mode. Default is 1000000.                                                                                                                            |
 
 
+### ZoneMap Options
+
+For the `zonemap` method, the following option is supported:
+
+| Option          | Type | Description                                       |
+|-----------------|------|---------------------------------------------------|
+| `rows_per_zone` | Long | Approximate number of rows per zone. Default 8192. |
+
+ZoneMap is single-column only — passing more than one column raises an `IllegalArgumentException` at plan time.
+
 ### FTS Options
 
 For the `fts` method, the following options are required:
@@ -101,6 +112,30 @@ Create an index and specify the `zone_size` for the B-tree:
     ALTER TABLE lance.db.users CREATE INDEX idx_id_zoned USING btree (id) WITH (zone_size = 2048);
     ```
 
+### ZoneMap Index
+
+Create a zonemap index for lightweight fragment pruning on a scalar column:
+
+=== "SQL"
+    ```sql
+    ALTER TABLE lance.db.events CREATE INDEX idx_ts_zm USING zonemap (event_ts);
+    ```
+
+With a custom zone size:
+
+=== "SQL"
+    ```sql
+    ALTER TABLE lance.db.events CREATE INDEX idx_ts_zm USING zonemap (event_ts) WITH (rows_per_zone = 2048);
+    ```
+
+The build path is configurable via SparkConf:
+
+| SparkConf key                              | Default | Effect                                                                                                                                                       |
+|--------------------------------------------|---------|--------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| `spark.lance.zonemap.consolidate.enabled`  | `false` | When `false`, each fragment is indexed by its own Spark task and committed as a separate IndexMetadata segment under a shared name. When `true`, workers compute per-fragment zone batches in memory and the driver merges them into one consolidated segment covering every fragment. |
+
+The default produces a manifest entry per fragment (one IndexMetadata per `<uuid>/zonemap.lance`). The opt-in consolidated mode produces a single entry covering every fragment — fewer manifest entries and lower Lance file-header overhead at the cost of routing per-fragment batches through the driver.
+
 ### Full-Text Search Index
 
 Create an FTS index on a text column:
@@ -139,11 +174,12 @@ Consider creating an index when:
 
 The `CREATE INDEX` command operates as follows:
 
-1.  **Distributed Index Building**: For each fragment in the Lance dataset, a separate task is launched to build an index on the specified column(s).
-2.  **Metadata Merging**: Once all per-fragment indexes are built, their metadata is collected and merged.
-3.  **Transactional Commit**: A new table version is committed with the new index information. The operation is atomic and ensures that concurrent reads are not affected.
+1.  **Distributed Index Building**: For each fragment in the Lance dataset, a separate task is launched to build an index on the specified column(s). For `zonemap` with `spark.lance.zonemap.consolidate.enabled=true`, the per-task work returns an Arrow batch to the driver instead of writing a file.
+2.  **Metadata Merging / Commit**: For `btree` and `fts`, per-fragment indexes are collected and merged before commit. For `zonemap` default, each fragment's segment is committed as its own IndexMetadata entry under the shared name. For `zonemap` consolidated, the driver writes one merged segment and commits a single IndexMetadata entry.
+3.  **Transactional Commit**: A new table version is committed with the new index information. The operation is atomic and ensures concurrent reads are not affected.
 
 ## Notes and Limitations
 
-- **Index Methods**: The `btree` and `fts` methods are supported for scalar index creation.
-- **Index Replacement**: If you create an index with the same name as an existing one, the old index will be replaced by the new one. This is because the underlying implementation uses `replace(true)`.
+- **Index Methods**: `btree`, `fts`, and `zonemap` are supported.
+- **ZoneMap is single-column**: Passing more than one column raises `IllegalArgumentException`.
+- **Index Replacement**: If you create an index with the same name as an existing one, the old index is replaced by the new one. For `zonemap` default, the replacement removes every previous segment under that name in the same transaction.

integration-tests/test_lance_spark.py

@@ -778,6 +778,76 @@ def test_create_btree_index_on_string(self, spark):
         assert len(query_result) == 3
         _assert_lance_index_metadata(spark, "default.employees", "idx_dept", "BTREE")
 
+    def test_create_zonemap_index(self, spark):
+        """Test CREATE INDEX with zonemap on a scalar column."""
+        spark.sql(
+            """
+            CREATE TABLE default.events (
+                id INT,
+                event_ts INT,
+                payload STRING
+            )
+            """
+        )
+
+        data = [(i, i * 100, f"p{i}") for i in range(100)]
+        df = spark.createDataFrame(data, ["id", "event_ts", "payload"])
+        df.writeTo("default.events").append()
+
+        result = spark.sql(
+            """
+            ALTER TABLE default.events
+            CREATE INDEX idx_ts_zm USING zonemap (event_ts) WITH (rows_per_zone = 16)
+            """
+        ).collect()
+        assert len(result) == 1
+        assert result[0][1] == "idx_ts_zm"
+
+        # SHOW INDEXES surfaces the zonemap-typed segment(s).
+        indexes = spark.sql("SHOW INDEXES IN default.events").collect()
+        zonemap_rows = [row for row in indexes if row["name"] == "idx_ts_zm"]
+        assert len(zonemap_rows) >= 1
+        assert zonemap_rows[0]["index_type"] == "zonemap"
+
+        # Indexed query still returns the correct row.
+        query_result = spark.sql(
+            "SELECT * FROM default.events WHERE event_ts = 5000"
+        ).collect()
+        assert len(query_result) == 1
+        assert query_result[0].id == 50
+
+    def test_create_zonemap_index_consolidated(self, spark):
+        """ZoneMap with spark.lance.zonemap.consolidate.enabled=true commits one segment."""
+        spark.sql(
+            """
+            CREATE TABLE default.metrics (
+                id INT,
+                bucket INT
+            )
+            """
+        )
+
+        data = [(i, i % 10) for i in range(80)]
+        df = spark.createDataFrame(data, ["id", "bucket"])
+        df.writeTo("default.metrics").append()
+
+        spark.conf.set("spark.lance.zonemap.consolidate.enabled", "true")
+        try:
+            spark.sql(
+                """
+                ALTER TABLE default.metrics
+                CREATE INDEX idx_bucket_zm USING zonemap (bucket)
+                """
+            ).collect()
+        finally:
+            spark.conf.unset("spark.lance.zonemap.consolidate.enabled")
+
+        indexes = spark.sql("SHOW INDEXES IN default.metrics").collect()
+        zonemap_rows = [row for row in indexes if row["name"] == "idx_bucket_zm"]
+        # Consolidated path commits exactly one IndexMetadata segment.
+        assert len(zonemap_rows) == 1
+        assert zonemap_rows[0]["index_type"] == "zonemap"
+
     def test_create_fts_index(self, spark):
         """Test CREATE INDEX with full-text search (FTS)."""
         spark.sql("""