feat(gluten): export a Lance fragment scan as an Arrow C stream

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.7.1"
+current_version = "0.8.0-beta.1"
 parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)(-(?P<pre_label>alpha|beta|rc)\\.(?P<pre_n>\\d+))?"
 serialize = [
     "{major}.{minor}.{patch}-{pre_label}.{pre_n}",

docs/src/config.md

@@ -579,6 +579,17 @@ The parent configuration effectively "anchors" your Spark catalog at a specific
 hierarchy, making the extra levels transparent to Spark users while maintaining compatibility with the underlying
 namespace implementation.
 
+## Branch Read Option
+
+Set `branch` to read the current head of a named branch. Do not set `version` on the same read.
+
+```python
+df = spark.read \
+    .format("lance") \
+    .option("branch", "audit") \
+    .load("/path/to/dataset.lance")
+```
+
 ## Memory Configuration
 
 Lance Spark uses Arrow for data transfer between native code and Spark, and maintains caches for improved performance.

docs/src/operations/dql/select.md

@@ -199,6 +199,94 @@ Use `VERSION AS OF` to query a specific version of the table:
     df.show();
     ```
 
+### Query by Tag
+
+Use `VERSION AS OF` with a quoted tag name to query the snapshot referenced by a tag:
+
+=== "SQL"
+    ```sql
+    -- Query the snapshot referenced by the release_candidate tag
+    SELECT * FROM users VERSION AS OF 'release_candidate';
+
+    -- Query specific columns from a tagged snapshot
+    SELECT id, name FROM users VERSION AS OF 'v1.0';
+    ```
+
+=== "Python"
+    ```python
+    # Query a tag using SQL
+    spark.sql("SELECT * FROM users VERSION AS OF 'release_candidate'").show()
+    ```
+
+=== "Scala"
+    ```scala
+    // Query a tag using SQL
+    spark.sql("SELECT * FROM users VERSION AS OF 'release_candidate'").show()
+    ```
+
+=== "Java"
+    ```java
+    // Query a tag using SQL
+    spark.sql("SELECT * FROM users VERSION AS OF 'release_candidate'").show();
+    ```
+
+!!! note
+    Tags whose names consist entirely of integer digits cannot be queried. Numeric values are
+    interpreted as table versions, even when quoted. For example, both `VERSION AS OF 123` and
+    `VERSION AS OF '123'` query table version 123 rather than a tag named `123`. Use a tag name that
+    contains at least one non-digit character.
+
+Tag queries are read-only. `UPDATE`, `DELETE`, `INSERT`, `MERGE INTO`, `ADD COLUMNS`, and
+`UPDATE COLUMNS` operations cannot target a tagged snapshot.
+
+### Query by Branch
+
+Read the current head of a named branch. Do not set `branch` and `version` on the same read.
+
+=== "SQL"
+    ```sql
+    SELECT * FROM catalog.db.users.branch_audit;
+    ```
+
+=== "Python"
+    ```python
+    audit = spark.read.option("branch", "audit").table("catalog.db.users")
+
+    audit_path = spark.read \
+        .format("lance") \
+        .option("branch", "audit") \
+        .load("/path/to/dataset.lance")
+    ```
+
+=== "Scala"
+    ```scala
+    val audit = spark.read.option("branch", "audit").table("catalog.db.users")
+
+    val auditPath = spark.read
+        .format("lance")
+        .option("branch", "audit")
+        .load("/path/to/dataset.lance")
+    ```
+
+=== "Java"
+    ```java
+    Dataset<Row> audit = spark.read()
+        .option("branch", "audit")
+        .table("catalog.db.users");
+
+    Dataset<Row> auditPath = spark.read()
+        .format("lance")
+        .option("branch", "audit")
+        .load("/path/to/dataset.lance");
+    ```
+
+If a table named `users.branch_audit` already exists, Spark reads that table. Otherwise the last
+segment is a branch on the parent table. Do not add `VERSION AS OF` or `TIMESTAMP AS OF`. Branch
+identifiers are read-only; mutating commands are rejected.
+
+`.option("branch").table(...)` applies the branch at scan time. Spark still analyzes with the table
+schema. Use `table.branch_name` when the branch schema differs from the table.
+
 ### Query by Timestamp
 
 Use `TIMESTAMP AS OF` to query the table as it existed at a specific point in time:
@@ -239,6 +327,7 @@ These options control how data is read from Lance datasets. They can be set usin
 | `batch_size`          | Integer | `8192`  | Number of rows to read per batch during scanning. Larger values may improve throughput but increase memory usage. |
 | `use_scalar_index`    | Boolean | `true`  | Whether to use scalar indices (e.g. btree) for filter acceleration during scanning.                               |
 | `version`             | Integer | Latest  | Specific dataset version to read. If not specified, reads the latest version.                                     |
+| `branch`              | String  | Main    | Named branch to read. Reads the current head. Do not set with `version`.                                          |
 | `block_size`          | Integer | -       | Block size in bytes for reading data.                                                                             |
 | `index_cache_size`    | Integer | -       | Size of the index cache in number of entries.                                                                     |
 | `metadata_cache_size` | Integer | -       | Size of the metadata cache in number of entries.                                                                  |

integration-tests/test_lance_spark.py

@@ -2702,6 +2702,49 @@ def test_version_as_of(self, spark):
         assert len(result) == 1
         assert result[0].id == 1
 
+    def test_tag_as_of_excludes_data_inserted_after_tag_creation(self, spark):
+        """Test that a tag remains on its snapshot after the main table advances."""
+        spark.sql("""
+            CREATE TABLE default.test_table (
+                id INT,
+                name STRING
+            )
+        """)
+        spark.sql("""
+            INSERT INTO default.test_table VALUES
+            (1, 'before_tag_1'),
+            (2, 'before_tag_2')
+        """)
+        spark.sql("ALTER TABLE default.test_table CREATE TAG stable")
+
+        spark.sql("""
+            INSERT INTO default.test_table VALUES
+            (3, 'after_tag_1'),
+            (4, 'after_tag_2')
+        """)
+
+        tagged = spark.sql("""
+            SELECT id, name
+            FROM default.test_table VERSION AS OF 'stable'
+            ORDER BY id
+        """).collect()
+        current = spark.sql("""
+            SELECT id, name
+            FROM default.test_table
+            ORDER BY id
+        """).collect()
+
+        assert [(row.id, row.name) for row in tagged] == [
+            (1, "before_tag_1"),
+            (2, "before_tag_2"),
+        ]
+        assert [(row.id, row.name) for row in current] == [
+            (1, "before_tag_1"),
+            (2, "before_tag_2"),
+            (3, "after_tag_1"),
+            (4, "after_tag_2"),
+        ]
+
     @requires_update_or_merge
     def test_version_as_of_after_update(self, spark):
         """Test VERSION AS OF returns data before an update."""
@@ -2761,6 +2804,103 @@ def test_version_as_of_after_delete(self, spark):
         assert len(result) == 3
 
 
+class TestDQLBranchRead:
+    def test_branch_identifier_matches_option_and_path(self, spark):
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql(
+            "INSERT INTO default.test_table VALUES (1, 'a'), (2, 'b')"
+        )
+        expected = [(1, "a"), (2, "b")]
+        spark.sql(
+            "ALTER TABLE default.test_table CREATE BRANCH test_branch"
+        )
+        spark.sql(
+            "INSERT INTO default.test_table VALUES (3, 'c'), (4, 'd')"
+        )
+
+        identifier = spark.sql(
+            "SELECT * FROM default.test_table.branch_test_branch ORDER BY id"
+        ).collect()
+        option_table = (
+            spark.read.option("branch", "test_branch")
+            .table("default.test_table")
+            .orderBy("id")
+            .collect()
+        )
+        option_path = (
+            spark.read.format("lance")
+            .option("branch", "test_branch")
+            .load(_table_location(spark, "default.test_table"))
+            .orderBy("id")
+            .collect()
+        )
+        main = spark.sql(
+            "SELECT * FROM default.test_table ORDER BY id"
+        ).collect()
+
+        assert [(row.id, row.name) for row in identifier] == expected
+        assert [(row.id, row.name) for row in option_table] == expected
+        assert [(row.id, row.name) for row in option_path] == expected
+        assert [(row.id, row.name) for row in main] == expected + [(3, "c"), (4, "d")]
+
+    def test_branch_identifier_rejects_as_of_and_conflicting_options(self, spark):
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'main')")
+        spark.sql("ALTER TABLE default.test_table CREATE BRANCH audit")
+
+        with pytest.raises(Exception, match="Cannot combine"):
+            spark.sql(
+                "SELECT * FROM default.test_table.branch_audit VERSION AS OF 1"
+            ).collect()
+        with pytest.raises(Exception, match="Cannot combine"):
+            spark.sql(
+                "SELECT * FROM default.test_table.branch_audit TIMESTAMP AS OF now()"
+            ).collect()
+        with pytest.raises(Exception):
+            spark.read.option("branch", "audit").option("version", "1").table(
+                "default.test_table"
+            ).collect()
+        with pytest.raises(Exception, match="no_such_branch"):
+            spark.read.option("branch", "no_such_branch").table(
+                "default.test_table"
+            ).collect()
+
+    def test_branch_identifier_is_read_only(self, spark):
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'main')")
+        spark.sql("ALTER TABLE default.test_table CREATE BRANCH audit")
+
+        with pytest.raises(Exception):
+            spark.sql(
+                "INSERT INTO default.test_table.branch_audit VALUES (2, 'branch')"
+            ).collect()
+
+        assert spark.table("default.test_table").count() == 1
+        assert spark.table("default.test_table.branch_audit").count() == 1
+
+    def test_existing_table_wins_over_branch_identifier(self, spark):
+        if getattr(spark, "_lance_backend", None) == "glue":
+            pytest.skip("Glue table identifiers are database.table")
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'branch_row')")
+        spark.sql("ALTER TABLE default.test_table CREATE BRANCH audit")
+        spark.sql(
+            "CREATE TABLE default.test_table.branch_audit (id INT, name STRING)"
+        )
+        spark.sql(
+            "INSERT INTO default.test_table.branch_audit VALUES (99, 'literal')"
+        )
+
+        rows = spark.table("default.test_table.branch_audit").collect()
+        assert [(row.id, row.name) for row in rows] == [(99, "literal")]
+        assert [
+            row.id
+            for row in spark.read.option("branch", "audit")
+            .table("default.test_table")
+            .collect()
+        ] == [1]
+
+
 @requires_update_or_merge
 class TestDMLMergeDelete:
     """Test MERGE INTO with WHEN MATCHED THEN DELETE."""

lance-spark-3.4_2.12/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.lance</groupId>
         <artifactId>lance-spark-root</artifactId>
-        <version>0.7.1</version>
+        <version>0.8.0-beta.1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

lance-spark-3.4_2.12/src/main/java/org/lance/spark/LancePositionDeltaDataset.java

@@ -74,6 +74,7 @@ public LancePositionDeltaDataset(
   @Override
   public RowLevelOperationBuilder newRowLevelOperationBuilder(
       RowLevelOperationInfo rowLevelOperationInfo) {
+    ensureWritable();
     return new LanceRowLevelOperationBuilder(
         rowLevelOperationInfo.command(),
         sparkSchema,

lance-spark-3.4_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java

@@ -22,6 +22,7 @@
 import org.lance.namespace.LanceNamespace;
 import org.lance.operation.Update;
 import org.lance.spark.LanceConstant;
+import org.lance.spark.LanceRef;
 import org.lance.spark.LanceRuntime;
 import org.lance.spark.LanceSparkWriteOptions;
 import org.lance.spark.function.LanceFragmentIdWithDefaultFunction;
@@ -95,9 +96,8 @@ public SparkPositionDeltaWrite(
       List<String> tableId) {
     this.sparkSchema = sparkSchema;
     try (Dataset ds = Utils.openDatasetBuilder(writeOptions).build()) {
-      this.writeOptions = writeOptions.withVersion(ds.version());
-      logger.debug(
-          "Resolved dataset version for position delta write: {}", this.writeOptions.getVersion());
+      this.writeOptions = writeOptions.withRef(LanceRef.ofMain(ds.version()));
+      logger.debug("Resolved dataset ref for position delta write: {}", this.writeOptions.getRef());
     }
     this.initialStorageOptions = initialStorageOptions;
     this.namespaceImpl = namespaceImpl;
@@ -173,8 +173,10 @@ public void commit(WriterCommitMessage[] messages) {
 
       long version =
           Objects.requireNonNull(
-              writeOptions.getVersion(),
-              "version must be set (resolved in SparkPositionDeltaWrite constructor)");
+                  writeOptions.getRef(),
+                  "ref must be set (resolved in SparkPositionDeltaWrite constructor)")
+              .getVersionNumber()
+              .get();
       try (Dataset dataset = Utils.openDatasetBuilder(writeOptions).build()) {
         // Parallel stream is safe: each deleteRows() operates on an independent
         // FileFragment value writing to a distinct object store path (see lance-core).

lance-spark-3.4_2.12/src/test/java/org/lance/spark/tag/TagDQLTest.java

@@ -0,0 +1,16 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.lance.spark.tag;
+
+public class TagDQLTest extends BaseTagDQLTest {}

lance-spark-3.4_2.13/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.lance</groupId>
         <artifactId>lance-spark-root</artifactId>
-        <version>0.7.1</version>
+        <version>0.8.0-beta.1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

lance-spark-3.5_2.12/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.lance</groupId>
         <artifactId>lance-spark-root</artifactId>
-        <version>0.7.1</version>
+        <version>0.8.0-beta.1</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

lance-spark-3.5_2.12/src/main/java/org/lance/spark/LancePositionDeltaDataset.java

@@ -74,6 +74,7 @@ public LancePositionDeltaDataset(
   @Override
   public RowLevelOperationBuilder newRowLevelOperationBuilder(
       RowLevelOperationInfo rowLevelOperationInfo) {
+    ensureWritable();
     return new LanceRowLevelOperationBuilder(
         rowLevelOperationInfo.command(),
         sparkSchema,