test: add unit tests for read/write pipeline and schema converter (#382)

## Summary

Add 52 unit tests across 6 new test classes covering previously untested
core components in the read pipeline, write pipeline, and schema
conversion layer.

## Motivation

Several critical classes in the read and write pipelines had zero direct
unit test coverage, relying entirely on integration tests for
validation. This makes it harder to catch regressions in query
optimization (filter/limit/topN push-down), write mode switching (append
vs overwrite), staged commit lifecycle, and schema type mapping.

## New Test Classes

### Read Pipeline

- **LanceScanBuilderTest** (19 tests) — Tests all 6 Spark push-down
interfaces:
  - `pruneColumns`: schema update and empty schema
- `pushFilters`: all-supported filters, mixed supported/unsupported
split, empty array, disabled via `pushDownFilters=false` config, skips
`Array<Struct>` schemas
([lance#3578](lance-format/lance#3578))
- `pushLimit`, `pushOffset` (rejects multi-fragment),
`isPartiallyPushed`
- `pushTopN`: enabled/disabled by config, rejects non-`FieldReference`
expressions
- `pushAggregation`: `COUNT(*)` from metadata vs scanner fallback with
filters, rejects `GROUP BY` and non-`COUNT` aggregations
- `build()`: returns `LanceScan` vs `LanceLocalScan` for metadata-based
`COUNT(*)`

- **LanceScanTest** (5 tests) — Tests scan execution:
- `readSchema()` returns original schema or count schema when
aggregation is pushed
- `planInputPartitions()` returns non-empty partitions, propagates
filters and limit to `LanceInputPartition`

- **LanceSplitTest** (3 tests) — Tests partition planning:
- `planScan()` returns splits with resolved version for snapshot
isolation
  - Each split maps to a single fragment
  - Deprecated `generateLanceSplits()` backward compatibility

### Write Pipeline

- **SparkWriteTest** (4 tests) — Tests write interface contract:
- `build()` returns `SparkWrite`, `toBatch()` returns `LanceBatchWrite`
  - `toStreaming()` throws `UnsupportedOperationException`
  - `truncate()` switches write mode from `APPEND` to `OVERWRITE`

- **StagedCommitTest** (4 tests) — Tests staged commit lifecycle with
real datasets:
  - Commit new table creates dataset on disk
  - Commit existing table overwrites with empty fragments
- Abort new table (without namespace) and existing table release
resources

### Schema Conversion

- **SchemaConverterTest** (17 tests) — Tests Spark-to-Arrow type
mapping:
- `toJsonArrowSchema`: 11 primitive types, nullability, `ArrayType`,
`FixedSizeList` (with metadata), `StructType`, `MapType`, `DecimalType`,
empty schema
- `processSchemaWithProperties`: vector/blob/large-varchar metadata
injection, null/empty properties passthrough, type validation errors
(non-array vector, non-float element, non-binary blob, non-string
large-varchar)

Author: LuciferYang

lance-spark-base_2.12/src/test/java/org/lance/spark/read/LanceScanBuilderTest.java

@@ -0,0 +1,328 @@
+/*
+ * 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.read;
+
+import org.lance.spark.LanceSparkReadOptions;
+import org.lance.spark.TestUtils;
+
+import org.apache.spark.sql.connector.expressions.Expression;
+import org.apache.spark.sql.connector.expressions.FieldReference;
+import org.apache.spark.sql.connector.expressions.NullOrdering;
+import org.apache.spark.sql.connector.expressions.SortDirection;
+import org.apache.spark.sql.connector.expressions.SortOrder;
+import org.apache.spark.sql.connector.expressions.aggregate.AggregateFunc;
+import org.apache.spark.sql.connector.expressions.aggregate.Aggregation;
+import org.apache.spark.sql.connector.expressions.aggregate.CountStar;
+import org.apache.spark.sql.connector.expressions.aggregate.Sum;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.sources.Filter;
+import org.apache.spark.sql.sources.GreaterThan;
+import org.apache.spark.sql.sources.IsNotNull;
+import org.apache.spark.sql.sources.LessThan;
+import org.apache.spark.sql.sources.StringContains;
+import org.apache.spark.sql.types.ArrayType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class LanceScanBuilderTest {
+
+  private static final StructType TEST_SCHEMA = TestUtils.TestTable1Config.schema;
+
+  private LanceScanBuilder createBuilder() {
+    return new LanceScanBuilder(
+        TEST_SCHEMA,
+        TestUtils.TestTable1Config.readOptions,
+        Collections.emptyMap(),
+        null,
+        Collections.emptyMap());
+  }
+
+  // --- pruneColumns ---
+
+  @Test
+  public void testPruneColumnsUpdatesSchema() {
+    LanceScanBuilder builder = createBuilder();
+    StructType requiredSchema =
+        new StructType(
+            new StructField[] {
+              DataTypes.createStructField("x", DataTypes.LongType, true),
+            });
+    builder.pruneColumns(requiredSchema);
+    Scan scan = builder.build();
+    assertEquals(requiredSchema, scan.readSchema());
+  }
+
+  @Test
+  public void testPruneColumnsToEmptySchema() {
+    LanceScanBuilder builder = createBuilder();
+    StructType emptySchema = new StructType();
+    builder.pruneColumns(emptySchema);
+    Scan scan = builder.build();
+    assertEquals(emptySchema, scan.readSchema());
+  }
+
+  // --- pushFilters ---
+
+  @Test
+  public void testPushFiltersAllSupported() {
+    LanceScanBuilder builder = createBuilder();
+    Filter[] filters =
+        new Filter[] {
+          new GreaterThan("x", 1L), new LessThan("y", 10L), new IsNotNull("b"),
+        };
+    Filter[] postScanFilters = builder.pushFilters(filters);
+    assertEquals(0, postScanFilters.length);
+    assertEquals(3, builder.pushedFilters().length);
+  }
+
+  @Test
+  public void testPushFiltersMixedSupportedAndUnsupported() {
+    LanceScanBuilder builder = createBuilder();
+    // StringContains is not supported for push-down
+    Filter[] filters =
+        new Filter[] {
+          new GreaterThan("x", 1L), new StringContains("b", "test"),
+        };
+    Filter[] postScanFilters = builder.pushFilters(filters);
+    assertEquals(1, postScanFilters.length);
+    assertInstanceOf(StringContains.class, postScanFilters[0]);
+    assertEquals(1, builder.pushedFilters().length);
+    assertInstanceOf(GreaterThan.class, builder.pushedFilters()[0]);
+  }
+
+  @Test
+  public void testPushFiltersEmptyArray() {
+    LanceScanBuilder builder = createBuilder();
+    Filter[] result = builder.pushFilters(new Filter[0]);
+    assertEquals(0, result.length);
+    assertEquals(0, builder.pushedFilters().length);
+  }
+
+  @Test
+  public void testPushFiltersDisabledByConfig() {
+    LanceSparkReadOptions options =
+        LanceSparkReadOptions.from(
+            Collections.singletonMap(LanceSparkReadOptions.CONFIG_PUSH_DOWN_FILTERS, "false"),
+            TestUtils.TestTable1Config.datasetUri);
+    LanceScanBuilder builder =
+        new LanceScanBuilder(TEST_SCHEMA, options, Collections.emptyMap(), null, null);
+    Filter[] filters = new Filter[] {new GreaterThan("x", 1L)};
+    Filter[] result = builder.pushFilters(filters);
+    assertEquals(1, result.length);
+    assertEquals(0, builder.pushedFilters().length);
+  }
+
+  @Test
+  public void testPushFiltersSkipsNestedArrayOfStruct() {
+    // Workaround for https://github.com/lance-format/lance/issues/3578
+    StructType nestedSchema =
+        new StructType(
+            new StructField[] {
+              DataTypes.createStructField("id", DataTypes.LongType, true),
+              DataTypes.createStructField(
+                  "items",
+                  new ArrayType(
+                      new StructType(
+                          new StructField[] {
+                            DataTypes.createStructField("name", DataTypes.StringType, true)
+                          }),
+                      true),
+                  true),
+            });
+    LanceScanBuilder builder =
+        new LanceScanBuilder(
+            nestedSchema,
+            TestUtils.TestTable1Config.readOptions,
+            Collections.emptyMap(),
+            null,
+            Collections.emptyMap());
+    Filter[] filters = new Filter[] {new GreaterThan("id", 1L)};
+    Filter[] result = builder.pushFilters(filters);
+    assertEquals(1, result.length);
+    assertEquals(0, builder.pushedFilters().length);
+  }
+
+  // --- pushLimit ---
+
+  @Test
+  public void testPushLimitAlwaysSucceeds() {
+    LanceScanBuilder builder = createBuilder();
+    assertTrue(builder.pushLimit(100));
+  }
+
+  // --- pushOffset ---
+
+  @Test
+  public void testPushOffsetRejectsMultiFragmentDataset() {
+    // TestTable1 has 2 fragments, so offset cannot be pushed
+    LanceScanBuilder builder = createBuilder();
+    assertFalse(builder.pushOffset(10));
+  }
+
+  @Test
+  public void testIsPartiallyPushedAlwaysTrue() {
+    LanceScanBuilder builder = createBuilder();
+    assertTrue(builder.isPartiallyPushed());
+  }
+
+  // --- pushTopN ---
+
+  @Test
+  public void testPushTopNEnabledByDefault() {
+    LanceScanBuilder builder = createBuilder();
+    SortOrder order = new TestSortOrder("x", SortDirection.ASCENDING, NullOrdering.NULLS_FIRST);
+    assertTrue(builder.pushTopN(new SortOrder[] {order}, 10));
+  }
+
+  @Test
+  public void testPushTopNDisabledByConfig() {
+    LanceSparkReadOptions options =
+        LanceSparkReadOptions.from(
+            Collections.singletonMap(LanceSparkReadOptions.CONFIG_TOP_N_PUSH_DOWN, "false"),
+            TestUtils.TestTable1Config.datasetUri);
+    LanceScanBuilder builder =
+        new LanceScanBuilder(TEST_SCHEMA, options, Collections.emptyMap(), null, null);
+    SortOrder order = new TestSortOrder("x", SortDirection.ASCENDING, NullOrdering.NULLS_FIRST);
+    assertFalse(builder.pushTopN(new SortOrder[] {order}, 10));
+  }
+
+  @Test
+  public void testPushTopNRejectsNonFieldReferenceExpression() {
+    LanceScanBuilder builder = createBuilder();
+    // A SortOrder whose expression is not a FieldReference should be rejected
+    SortOrder nonFieldOrder =
+        new SortOrder() {
+          @Override
+          public Expression expression() {
+            return new Expression() {
+              @Override
+              public Expression[] children() {
+                return new Expression[0];
+              }
+
+              @Override
+              public String toString() {
+                return "custom_expression";
+              }
+            };
+          }
+
+          @Override
+          public SortDirection direction() {
+            return SortDirection.ASCENDING;
+          }
+
+          @Override
+          public NullOrdering nullOrdering() {
+            return NullOrdering.NULLS_FIRST;
+          }
+        };
+    assertFalse(builder.pushTopN(new SortOrder[] {nonFieldOrder}, 10));
+  }
+
+  // --- pushAggregation ---
+
+  @Test
+  public void testPushAggregationCountStarFromMetadata() {
+    LanceScanBuilder builder = createBuilder();
+    Aggregation countStar =
+        new Aggregation(new AggregateFunc[] {new CountStar()}, new Expression[] {});
+    assertTrue(builder.pushAggregation(countStar));
+  }
+
+  @Test
+  public void testPushAggregationCountStarWithFiltersFallsBackToScanner() {
+    LanceScanBuilder builder = createBuilder();
+    builder.pushFilters(new Filter[] {new GreaterThan("x", 0L)});
+    Aggregation countStar =
+        new Aggregation(new AggregateFunc[] {new CountStar()}, new Expression[] {});
+    // With pushed filters, metadata count cannot be used; falls back to scanner-based count
+    assertTrue(builder.pushAggregation(countStar));
+  }
+
+  @Test
+  public void testPushAggregationRejectsGroupBy() {
+    LanceScanBuilder builder = createBuilder();
+    Aggregation groupedAgg =
+        new Aggregation(
+            new AggregateFunc[] {new CountStar()}, new Expression[] {FieldReference.apply("x")});
+    assertFalse(builder.pushAggregation(groupedAgg));
+  }
+
+  @Test
+  public void testPushAggregationRejectsNonCountStar() {
+    LanceScanBuilder builder = createBuilder();
+    Aggregation sumAgg =
+        new Aggregation(
+            new AggregateFunc[] {new Sum(FieldReference.apply("x"), false)}, new Expression[] {});
+    assertFalse(builder.pushAggregation(sumAgg));
+  }
+
+  // --- build ---
+
+  @Test
+  public void testBuildReturnsLanceScan() {
+    LanceScanBuilder builder = createBuilder();
+    Scan scan = builder.build();
+    assertNotNull(scan);
+    assertInstanceOf(LanceScan.class, scan);
+    assertEquals(TEST_SCHEMA, scan.readSchema());
+  }
+
+  @Test
+  public void testBuildWithCountStarReturnsLocalScan() {
+    LanceScanBuilder builder = createBuilder();
+    Aggregation countStar =
+        new Aggregation(new AggregateFunc[] {new CountStar()}, new Expression[] {});
+    builder.pushAggregation(countStar);
+    Scan scan = builder.build();
+    // Metadata-based COUNT(*) without filters returns LanceLocalScan
+    assertNotNull(scan);
+    assertInstanceOf(LanceLocalScan.class, scan);
+  }
+
+  /** Minimal SortOrder implementation for testing pushTopN. */
+  private static class TestSortOrder implements SortOrder {
+    private final String columnName;
+    private final SortDirection direction;
+    private final NullOrdering nullOrdering;
+
+    TestSortOrder(String columnName, SortDirection direction, NullOrdering nullOrdering) {
+      this.columnName = columnName;
+      this.direction = direction;
+      this.nullOrdering = nullOrdering;
+    }
+
+    @Override
+    public Expression expression() {
+      return FieldReference.apply(columnName);
+    }
+
+    @Override
+    public SortDirection direction() {
+      return direction;
+    }
+
+    @Override
+    public NullOrdering nullOrdering() {
+      return nullOrdering;
+    }
+  }
+}