[VL] Adding example customized operator

backends-velox/src/main/scala/org/apache/gluten/execution/GlutenStrideExecTransformer.scala

@@ -0,0 +1,112 @@
+/*
+ * 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.gluten.execution
+
+import org.apache.gluten.metrics.MetricsUpdater
+import org.apache.gluten.substrait.SubstraitContext
+import org.apache.gluten.substrait.rel.{RelBuilder, RelNode}
+
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.SparkPlan
+
+import scala.collection.JavaConverters._
+
+/**
+ * Gluten-native "stride" operator: outputs every N-th row from each input batch.
+ *
+ * Row indices within each batch are 0-based; index 0 is always included. The counter resets per
+ * batch, so the operator is completely stateless and parallelism-friendly.
+ *
+ * This is a self-contained example of the Gluten custom-operator mechanism. It does NOT correspond
+ * to any existing Spark operator -- it exists purely to demonstrate how to add a backend-specific
+ * native operator without requiring a Velox upstream contribution.
+ *
+ * ==End-to-end flow==
+ * {{{
+ *   GlutenStrideExecTransformer(stride=3, child=scanPlan)
+ *     --doTransform-->  GlutenStrideRelNode(stride=3)
+ *       --toProtobuf--> FetchRel { offset=3, advanced_extension.optimization="isGlutenStride=1" }
+ *         --JNI-->      SubstraitToVeloxPlanConverter::toVeloxPlan(FetchRel&)
+ *           --builds--> GlutenStrideNode(stride=3, child)
+ *             --exec--> GlutenStrideOperator: keeps rows 0, 3, 6, 9, ...
+ * }}}
+ *
+ * ==Usage==
+ * Instantiate directly in a unit test or wire it into a custom offload rule:
+ * {{{
+ *   val strider = GlutenStrideExecTransformer(stride = 3L, child = childPlan)
+ * }}}
+ */
+case class GlutenStrideExecTransformer(stride: Long, child: SparkPlan)
+  extends UnaryTransformSupport {
+
+  require(stride >= 1L, s"stride must be >= 1, got $stride")
+
+  // -------------------------------------------------------------------------
+  // SparkPlan identity
+  // -------------------------------------------------------------------------
+
+  override def output: Seq[Attribute] = child.output
+
+  override def metricsUpdater(): MetricsUpdater = MetricsUpdater.None
+
+  override protected def withNewChildInternal(newChild: SparkPlan): GlutenStrideExecTransformer =
+    copy(child = newChild)
+
+  // -------------------------------------------------------------------------
+  // Validation
+  // -------------------------------------------------------------------------
+
+  override protected def doValidateInternal(): ValidationResult = {
+    val context = new SubstraitContext
+    val operatorId = context.nextOperatorId(this.nodeName)
+    val relNode = makeRelNode(context, operatorId, inputRelNode = null, validation = true)
+    doNativeValidation(context, relNode)
+  }
+
+  // -------------------------------------------------------------------------
+  // Transformation
+  // -------------------------------------------------------------------------
+
+  override protected def doTransform(context: SubstraitContext): TransformContext = {
+    val childCtx = child.asInstanceOf[TransformSupport].transform(context)
+    val operatorId = context.nextOperatorId(this.nodeName)
+    val relNode = makeRelNode(context, operatorId, inputRelNode = childCtx.root, validation = false)
+    TransformContext(output, relNode)
+  }
+
+  // -------------------------------------------------------------------------
+  // Private helpers
+  // -------------------------------------------------------------------------
+
+  private def makeRelNode(
+      context: SubstraitContext,
+      operatorId: Long,
+      inputRelNode: RelNode,
+      validation: Boolean): RelNode = {
+    if (validation) {
+      RelBuilder.makeGlutenStrideRel(
+        inputRelNode,
+        stride,
+        RelBuilder.createExtensionNode(output.asJava),
+        context,
+        operatorId)
+    } else {
+      RelBuilder.makeGlutenStrideRel(inputRelNode, stride, context, operatorId)
+    }
+  }
+}

backends-velox/src/test/scala/org/apache/gluten/execution/GlutenStrideExecSuite.scala

@@ -0,0 +1,169 @@
+/*
+ * 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.gluten.execution
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.execution.SparkPlan
+
+/**
+ * Tests for [[GlutenStrideExecTransformer]] — the example Gluten-native custom operator.
+ *
+ * The operator keeps every N-th row within each input batch (indices 0, stride, 2*stride, …).
+ * It has no Spark logical-plan equivalent, so it cannot be exercised via SQL. Instead the tests
+ * construct the physical plan node directly and drive execution via
+ * [[SparkPlan.executeCollect()]], which triggers the full Velox pipeline.
+ */
+class GlutenStrideExecSuite extends VeloxWholeStageTransformerSuite {
+
+  override protected val resourcePath: String = "N/A"
+  override protected val fileFormat: String = "N/A"
+
+  override protected def sparkConf: SparkConf =
+    super.sparkConf
+      .set("spark.memory.offHeap.size", "512m")
+      // Single partition keeps tests deterministic — no per-partition counter reset.
+      .set("spark.sql.shuffle.partitions", "1")
+      .set("spark.default.parallelism", "1")
+
+  // ---------------------------------------------------------------------------
+  // Helper: wrap `child` in a GlutenStrideExecTransformer, execute it, and
+  // return the collected Long values from the first (id) column.
+  // ---------------------------------------------------------------------------
+  private def strideIds(child: SparkPlan, stride: Long): Seq[Long] = {
+    val strider = GlutenStrideExecTransformer(stride = stride, child = child)
+    strider.executeCollect().map(_.getLong(0)).toSeq
+  }
+
+  // ---------------------------------------------------------------------------
+  // Correctness tests
+  // ---------------------------------------------------------------------------
+
+  test("stride=1 returns all rows unchanged") {
+    withTable("stride_t") {
+      spark.range(0, 6).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").queryExecution.executedPlan
+      val result = strideIds(child, stride = 1L)
+      assert(result.sorted == Seq(0L, 1L, 2L, 3L, 4L, 5L),
+        s"stride=1 should return all 6 rows, got: $result")
+    }
+  }
+
+  test("stride=2 returns every other row") {
+    withTable("stride_t") {
+      // Write exactly 5 rows so the stride-2 result is deterministic within one batch
+      spark.range(0, 5).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").orderBy("id").queryExecution.executedPlan
+      val result = strideIds(child, stride = 2L)
+      // Sorted input: 0,1,2,3,4 → indices 0,2,4 → values 0,2,4
+      assert(result == Seq(0L, 2L, 4L),
+        s"stride=2 over 5 rows should give [0,2,4], got: $result")
+    }
+  }
+
+  test("stride=3 returns every third row") {
+    withTable("stride_t") {
+      // 9 rows sorted → indices 0,3,6 → values 0,3,6
+      spark.range(0, 9).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").orderBy("id").queryExecution.executedPlan
+      val result = strideIds(child, stride = 3L)
+      assert(result == Seq(0L, 3L, 6L),
+        s"stride=3 over 9 rows should give [0,3,6], got: $result")
+    }
+  }
+
+  test("stride > row count returns only the first row") {
+    withTable("stride_t") {
+      spark.range(0, 5).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").orderBy("id").queryExecution.executedPlan
+      val result = strideIds(child, stride = 100L)
+      assert(result == Seq(0L),
+        s"stride > row count should keep only row at index 0, got: $result")
+    }
+  }
+
+  test("stride applied to single-row input returns that single row") {
+    withTable("stride_t") {
+      spark.range(42L, 43L).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").queryExecution.executedPlan
+      val result = strideIds(child, stride = 5L)
+      assert(result == Seq(42L),
+        s"single-row input should be returned unchanged, got: $result")
+    }
+  }
+
+  test("stride applied to empty input returns empty result") {
+    withTable("stride_t") {
+      spark.range(0L, 0L).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").queryExecution.executedPlan
+      val result = strideIds(child, stride = 2L)
+      assert(result.isEmpty,
+        s"empty input should produce empty output, got: $result")
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+  // Plan-structure test: GlutenStrideExecTransformer appears in the plan tree
+  // ---------------------------------------------------------------------------
+
+  test("GlutenStrideExecTransformer appears in the plan tree string") {
+    withTable("stride_t") {
+      spark.range(0, 4).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").queryExecution.executedPlan
+      val strider = GlutenStrideExecTransformer(stride = 2L, child = child)
+      val planStr = strider.treeString
+      assert(
+        planStr.contains("GlutenStride"),
+        s"Expected 'GlutenStride' in plan tree but got:\n$planStr")
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+  // Chained plan: GlutenStrideExecTransformer wrapping an already-filtered scan
+  // ---------------------------------------------------------------------------
+
+  test("stride after filter produces correct subset") {
+    withTable("stride_t") {
+      // rows 0..9 → filter keeps even rows: 0,2,4,6,8 → stride=2 → indices 0,2,4 → values 0,4,8
+      spark.range(0, 10).write.format("parquet").saveAsTable("stride_t")
+      val filteredPlan = spark
+        .table("stride_t")
+        .filter("id % 2 = 0")
+        .orderBy("id")
+        .queryExecution
+        .executedPlan
+      val result = strideIds(filteredPlan, stride = 2L)
+      // Even values sorted: 0,2,4,6,8 → keep indices 0,2,4 → 0,4,8
+      assert(result == Seq(0L, 4L, 8L),
+        s"stride=2 after even-filter mismatch: $result")
+    }
+  }
+
+  // ---------------------------------------------------------------------------
+  // Argument validation
+  // ---------------------------------------------------------------------------
+
+  test("stride=0 is rejected at construction time") {
+    withTable("stride_t") {
+      spark.range(0, 3).write.format("parquet").saveAsTable("stride_t")
+      val child = spark.table("stride_t").queryExecution.executedPlan
+      val ex = intercept[IllegalArgumentException] {
+        GlutenStrideExecTransformer(stride = 0L, child = child)
+      }
+      assert(ex.getMessage.contains("stride"), s"Unexpected error message: ${ex.getMessage}")
+    }
+  }
+}

cpp/velox/CMakeLists.txt

@@ -178,6 +178,7 @@ set(VELOX_SRCS
     operators/functions/delta/DeltaBitmapAggregator.cc
     operators/functions/RowConstructorWithNull.cc
     operators/functions/SparkExprToSubfieldFilterParser.cc
+    operators/plannodes/GlutenStrideNode.cc
     operators/plannodes/RowVectorStream.cc
     operators/hashjoin/HashTableBuilder.cc
     operators/hashjoin/HashTableSerializer.cc

cpp/velox/compute/VeloxBackend.cc

@@ -47,6 +47,7 @@
 #include "jni/JniFileSystem.h"
 #include "memory/GlutenBufferedInputBuilder.h"
 #include "operators/functions/SparkExprToSubfieldFilterParser.h"
+#include "operators/plannodes/GlutenStrideNode.h"
 #include "operators/plannodes/RowVectorStream.h"
 #include "shuffle/ArrowShuffleDictionaryWriter.h"
 #include "udf/UdfLoader.h"
@@ -228,6 +229,9 @@ void VeloxBackend::init(
   }
 #endif
 
+  // Register Gluten-native custom operator translators.
+  facebook::velox::exec::Operator::registerOperator(std::make_unique<GlutenStrideTranslator>());
+
   const int32_t numTaskSlotsPerExecutor = [&]() {
     if (!backendConf_->valueExists(kNumTaskSlotsPerExecutor)) {
       LOG(WARNING) << kNumTaskSlotsPerExecutor << " is not set. Falling back to 1.";

cpp/velox/operators/plannodes/GlutenStrideNode.cc

@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+#include "GlutenStrideNode.h"
+
+#include "velox/exec/OperatorUtils.h"
+
+namespace gluten {
+
+GlutenStrideOperator::GlutenStrideOperator(
+    int32_t operatorId,
+    facebook::velox::exec::DriverCtx* driverCtx,
+    std::shared_ptr<const GlutenStrideNode> node)
+    : facebook::velox::exec::Operator(driverCtx, node->outputType(), operatorId, node->id(), "GlutenStride"),
+      stride_(node->stride()) {}
+
+facebook::velox::RowVectorPtr GlutenStrideOperator::getOutput() {
+  if (!input_) {
+    return nullptr;
+  }
+  auto inputBatch = std::move(input_);
+  const auto numRows = static_cast<int64_t>(inputBatch->size());
+
+  // Count how many rows we will keep: rows at indices 0, stride_, 2*stride_, ...
+  int64_t numSelected = 0;
+  for (int64_t i = 0; i < numRows; i += stride_) {
+    ++numSelected;
+  }
+
+  if (numSelected == numRows) {
+    // stride == 1: pass every row through unchanged.
+    return inputBatch;
+  }
+  if (numSelected == 0) {
+    return inputBatch; // empty batch — nothing to do
+  }
+
+  // Build an index buffer selecting rows 0, stride_, 2*stride_, ...
+  facebook::velox::BufferPtr indices =
+      facebook::velox::allocateIndices(static_cast<facebook::velox::vector_size_t>(numSelected), pool());
+  auto* rawIndices = indices->asMutable<facebook::velox::vector_size_t>();
+  facebook::velox::vector_size_t idx = 0;
+  for (int64_t i = 0; i < numRows; i += stride_) {
+    rawIndices[idx++] = static_cast<facebook::velox::vector_size_t>(i);
+  }
+
+  return facebook::velox::exec::wrap(
+      static_cast<facebook::velox::vector_size_t>(numSelected), std::move(indices), inputBatch);
+}
+
+} // namespace gluten