[VL] Add min/max partition stats to columnar InMemoryRelation cache for partition pruning

backends-velox/src/test/scala/org/apache/spark/sql/execution/ColumnarCacheShipBlockerMarshalSuite.scala

@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.spark.sql.execution
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.sql.types.{Decimal, DecimalType, StringType, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.Arrays
+
+/**
+ * Ship-blocker acceptance tests for the non-BIGINT marshal paths (Decimal short/long, String, Float
+ * / Double NaN guard). Guards against silent corruption (wrong Decimal scale, wrong UTF-8 byte
+ * order, lost BigInteger sign) on round-trip.
+ */
+class ColumnarCacheShipBlockerMarshalSuite extends AnyFunSuite {
+
+  test("Decimal(10, 2) round-trip preserves value") {
+    val lo = Decimal(BigDecimal("1.50"), 10, 2)
+    val hi = Decimal(BigDecimal("99.99"), 10, 2)
+    val stats: InternalRow = new GenericInternalRow(
+      Array[Any](lo, hi, 0, 100, 800L))
+    val schema = StructType(Seq(StructField("d", DecimalType(10, 2))))
+    val blob = CachedColumnarBatchKryoSerializer.serializeStats(stats, schema)
+    val read = CachedColumnarBatchKryoSerializer.deserializeStats(blob, schema)
+    val dt = DecimalType(10, 2)
+    val readLo = read.get(0, dt)
+    val readHi = read.get(1, dt)
+    assert(readLo == lo, s"lower bound corrupted: expected $lo got $readLo")
+    assert(readHi == hi, s"upper bound corrupted: expected $hi got $readHi")
+  }
+
+  test("Decimal(30, 5) round-trip preserves big value") {
+    val big = BigDecimal("12345678901234567890.12345")
+    val lo = Decimal(big.bigDecimal.negate(), 30, 5)
+    val hi = Decimal(big, 30, 5)
+    val stats: InternalRow = new GenericInternalRow(
+      Array[Any](lo, hi, 0, 100, 1600L))
+    val schema = StructType(Seq(StructField("d", DecimalType(30, 5))))
+    val blob = CachedColumnarBatchKryoSerializer.serializeStats(stats, schema)
+    val read = CachedColumnarBatchKryoSerializer.deserializeStats(blob, schema)
+    val dt = DecimalType(30, 5)
+    val readLo = read.get(0, dt)
+    val readHi = read.get(1, dt)
+    assert(readLo == lo, s"lower bound corrupted: expected $lo got $readLo")
+    assert(readHi == hi, s"upper bound corrupted: expected $hi got $readHi")
+  }
+
+  test("String byte-wise lex round-trip preserves UTF-8 bytes") {
+    val lo = UTF8String.fromString("apple")
+    // UTF-8 bytes for two CJK code points (U+4E2D U+6587) constructed from hex
+    // to keep this file ASCII-only (scalastyle nonascii filter).
+    val hi = UTF8String.fromBytes(Array[Byte](
+      0xe4.toByte,
+      0xb8.toByte,
+      0xad.toByte,
+      0xe6.toByte,
+      0x96.toByte,
+      0x87.toByte))
+    val stats: InternalRow = new GenericInternalRow(
+      Array[Any](lo, hi, 0, 100, 1024L))
+    val schema = StructType(Seq(StructField("s", StringType)))
+    val blob = CachedColumnarBatchKryoSerializer.serializeStats(stats, schema)
+    val read = CachedColumnarBatchKryoSerializer.deserializeStats(blob, schema)
+    val readLo = read.getUTF8String(0)
+    val readHi = read.getUTF8String(1)
+    assert(readLo == lo, s"lower bound corrupted: expected $lo got $readLo")
+    assert(readHi == hi, s"upper bound corrupted: expected $hi got $readHi")
+  }
+
+  test("String truncation to 256B widens upper bound monotonically") {
+    val loBytes = new Array[Byte](100)
+    Arrays.fill(loBytes, 'a'.toByte)
+    // Upper: 300 bytes of 'm' followed by trailing chars. Truncated prefix
+    // is 256 'm's (all 0x6d); +1 carry on last byte -> last byte = 0x6e.
+    val hiBytes = new Array[Byte](300)
+    Arrays.fill(hiBytes, 'm'.toByte)
+    val lo = UTF8String.fromBytes(loBytes)
+    val hi = UTF8String.fromBytes(hiBytes)
+    val stats: InternalRow = new GenericInternalRow(Array[Any](lo, hi, 0, 100, 50000L))
+    val schema = StructType(Seq(StructField("s", StringType)))
+    val blob = CachedColumnarBatchKryoSerializer.serializeStats(stats, schema)
+    val read = CachedColumnarBatchKryoSerializer.deserializeStats(blob, schema)
+    val readLo = read.getUTF8String(0)
+    val readHi = read.getUTF8String(1)
+    assert(readLo.numBytes() == 100, s"lower untruncated, got numBytes=${readLo.numBytes()}")
+    assert(readHi.numBytes() == 256, s"upper should truncate to 256B, got ${readHi.numBytes()}")
+    val readHiArr = readHi.getBytes
+    // First 255 bytes still 'm', last byte 'm'+1 = 'n' (0x6e) due to carry.
+    var i = 0
+    while (i < 255) {
+      assert(readHiArr(i) == 'm'.toByte, s"upper byte $i = ${readHiArr(i)}, expected 'm'")
+      i += 1
+    }
+    assert(
+      readHiArr(255) == 'n'.toByte,
+      s"upper byte 255 should be 'n' (carry), got ${readHiArr(255)}")
+  }
+
+  test("String carry overflow demotes column to unsupported") {
+    val loBytes = new Array[Byte](10)
+    Arrays.fill(loBytes, 0xff.toByte)
+    val hiBytes = new Array[Byte](300)
+    Arrays.fill(hiBytes, 0xff.toByte)
+    val lo = UTF8String.fromBytes(loBytes)
+    val hi = UTF8String.fromBytes(hiBytes)
+    val stats: InternalRow = new GenericInternalRow(Array[Any](lo, hi, 0, 100, 50000L))
+    val schema = StructType(Seq(StructField("s", StringType)))
+    val blob = CachedColumnarBatchKryoSerializer.serializeStats(stats, schema)
+    val read = CachedColumnarBatchKryoSerializer.deserializeStats(blob, schema)
+    assert(read.isNullAt(0), "lower bound must be null when carry overflows")
+    assert(read.isNullAt(1), "upper bound must be null when carry overflows")
+  }
+}

backends-velox/src/test/scala/org/apache/spark/sql/execution/ColumnarCachedBatchBuildFilterPruneSuite.scala

@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.spark.sql.execution
+
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, GenericInternalRow, Literal}
+import org.apache.spark.sql.columnar.CachedBatch
+import org.apache.spark.sql.types.LongType
+
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Tests for the lazy-split wrapper's stats!=null branch: batches whose [lowerBound, upperBound]
+ * does not cover the literal must be pruned; batches whose range covers it must be returned.
+ *
+ * Pure JVM. Uses GenericInternalRow with the vanilla 5-slot-per-col schema (lower, upper,
+ * nullCount, count, sizeInBytes).
+ */
+class ColumnarCachedBatchBuildFilterPruneSuite extends AnyFunSuite {
+
+  // Build a CachedColumnarBatch with BIGINT 1-col stats [lower, upper].
+  private def batchWithStats(numRows: Int, lower: Long, upper: Long): CachedColumnarBatch = {
+    val stats = new GenericInternalRow(Array[Any](lower, upper, 0, numRows, numRows * 8L))
+    CachedColumnarBatch(
+      numRows = numRows,
+      sizeInBytes = numRows * 8L,
+      bytes = Array.fill[Byte](numRows * 4)(0),
+      stats = stats)
+  }
+
+  test("EqualTo literal in [lower, upper] keeps the batch") {
+    val serializer = new ColumnarCachedBatchSerializer
+    val attr = AttributeReference("id", LongType, nullable = false)()
+    val predicate = EqualTo(attr, Literal(50L))
+    val filter = serializer.buildFilter(Seq(predicate), Seq(attr))
+
+    val batches: Iterator[CachedBatch] = Iterator(batchWithStats(10, 0L, 100L))
+    val result = filter(0, batches).toList
+
+    assert(result.length === 1, "batch with [0, 100] covers literal 50, must be kept")
+    assert(result.head.numRows === 10)
+  }
+
+  test("EqualTo literal outside [lower, upper] prunes the batch") {
+    val serializer = new ColumnarCachedBatchSerializer
+    val attr = AttributeReference("id", LongType, nullable = false)()
+    val predicate = EqualTo(attr, Literal(999L))
+    val filter = serializer.buildFilter(Seq(predicate), Seq(attr))
+
+    val batches: Iterator[CachedBatch] = Iterator(batchWithStats(10, 0L, 100L))
+    val result = filter(0, batches).toList
+
+    assert(result.length === 0, "batch with [0, 100] cannot contain 999, must be pruned")
+  }
+
+  test("mixed null/non-null stats: null through, non-null pruned by predicate") {
+    val serializer = new ColumnarCachedBatchSerializer
+    val attr = AttributeReference("id", LongType, nullable = false)()
+    val predicate = EqualTo(attr, Literal(50L))
+    val filter = serializer.buildFilter(Seq(predicate), Seq(attr))
+
+    val nullStats = CachedColumnarBatch(
+      numRows = 7,
+      sizeInBytes = 28L,
+      bytes = Array[Byte](9, 9),
+      stats = null)
+    val keptBatch = batchWithStats(10, 0L, 100L) // covers 50, kept
+    val prunedBatch = batchWithStats(5, 200L, 300L) // does not cover 50, pruned
+
+    val batches: Iterator[CachedBatch] = Iterator(nullStats, prunedBatch, keptBatch)
+    val result = filter(0, batches).toList
+
+    // Expected order: nullStats first (direct), then keptBatch.
+    // prunedBatch dropped by parent.
+    assert(
+      result.length === 2,
+      s"expected 2 (nullStats + keptBatch), got ${result.length}")
+    assert(
+      result.map(_.numRows) === Seq(7, 10),
+      "order: stats=null pass-through first, then stats-covers-literal kept")
+  }
+}

backends-velox/src/test/scala/org/apache/spark/sql/execution/ColumnarCachedBatchBuildFilterSuite.scala

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.spark.sql.execution
+
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, EqualTo, Literal}
+import org.apache.spark.sql.columnar.CachedBatch
+import org.apache.spark.sql.types.IntegerType
+
+import org.scalatest.funsuite.AnyFunSuite
+
+/**
+ * Tests for the buildFilter stats=null guard (lazy-split iterator wrapper). Without the wrapper,
+ * vanilla `SimpleMetricsCachedBatchSerializer.buildFilter` NPEs on `partitionFilter.eval(null)`
+ * (non-trivial predicates, both codegen and interpreted paths). A naive null-literal occupier row
+ * would fail differently and silently drop every batch.
+ *
+ * Pure JVM, no native lib. Drives buildFilter on a synthetic Iterator[CachedBatch] of
+ * CachedColumnarBatch with stats=null under a non-trivial predicate.
+ */
+class ColumnarCachedBatchBuildFilterSuite extends AnyFunSuite {
+
+  test("buildFilter must direct stats=null batches through under EqualTo predicate") {
+    val serializer = new ColumnarCachedBatchSerializer
+    val attr = AttributeReference("id", IntegerType, nullable = false)()
+    val predicate = EqualTo(attr, Literal(5))
+    val cachedAttributes = Seq(attr)
+
+    val filter = serializer.buildFilter(Seq(predicate), cachedAttributes)
+
+    // Three v1 binary batches: stats=null. A correct wrapper directs
+    // them all through (no pruning, since we have no stats to prune on).
+    val batches: Iterator[CachedBatch] = Iterator(
+      CachedColumnarBatch(
+        numRows = 3,
+        sizeInBytes = 12L,
+        bytes = Array[Byte](1, 2, 3),
+        stats = null),
+      CachedColumnarBatch(
+        numRows = 5,
+        sizeInBytes = 20L,
+        bytes = Array[Byte](4, 5),
+        stats = null),
+      CachedColumnarBatch(
+        numRows = 7,
+        sizeInBytes = 28L,
+        bytes = Array[Byte](6, 7, 8),
+        stats = null)
+    )
+
+    // GREEN expectation: all 3 batches returned (stats=null -> direct through).
+    // Without the wrapper: NPE thrown at partitionFilter.eval(null)
+    //   OR silent drop (if naive occupier-row placeholder was used).
+    val result = filter(0, batches).toList
+
+    assert(
+      result.length === 3,
+      "v1 binary (stats=null) batches must all be returned (lazy-split wrapper); " +
+        s"got ${result.length}")
+    assert(
+      result.map(_.numRows) === Seq(3, 5, 7),
+      "batch order must be preserved through wrapper")
+  }
+}