diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala index c017eb238be..67d1b0310d9 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/InputManager.scala @@ -153,4 +153,9 @@ class InputManager( inputBatch = batch currentInputIdx = -1 } + + // Marks the current batch fully consumed (hasUnfinishedInput becomes false). + def skipToEnd(): Unit = { + currentInputIdx = if (inputBatch == null) -1 else inputBatch.length - 1 + } } diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OutputManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OutputManager.scala index b15e8992546..09ac4cd5549 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OutputManager.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/messaginglayer/OutputManager.scala @@ -33,11 +33,16 @@ import org.apache.texera.amber.engine.architecture.messaginglayer.OutputManager. import org.apache.texera.amber.engine.architecture.sendsemantics.partitioners._ import org.apache.texera.amber.engine.architecture.sendsemantics.partitionings._ import org.apache.texera.amber.engine.architecture.worker.managers.{ + ArrowBatchWriteItem, OutputPortStorageWriterThread, - PortStorageWriterTerminateSignal + PortStorageWriterTerminateSignal, + RowWriteItem } import org.apache.texera.amber.engine.common.AmberLogging -import org.apache.texera.amber.util.VirtualIdentityUtils +import org.apache.texera.amber.engine.common.ambermessage.ColumnarFrame +import org.apache.texera.amber.util.{ArrowUtils, VirtualIdentityUtils} +import org.apache.arrow.memory.RootAllocator +import org.apache.arrow.vector.VectorSchemaRoot import java.net.URI import scala.collection.mutable @@ -204,6 +209,88 @@ class OutputManager( saveStateToStorageIfNeeded(state, loopCounter, loopStartId) } + // Allocator for the per-receiver sub-batches built during columnar shuffle. + // Created lazily on first shuffle emit, freed by closeColumnarResources(). + private var columnarEmitAllocator: RootAllocator = _ + private def emitAllocator(): RootAllocator = { + if (columnarEmitAllocator == null) columnarEmitAllocator = new RootAllocator() + columnarEmitAllocator + } + def closeColumnarResources(): Unit = { + if (columnarEmitAllocator != null) { + columnarEmitAllocator.close() + columnarEmitAllocator = null + } + } + + // Emit an Arrow batch downstream as ColumnarFrames, honoring each link's + // partitioning. Single-receiver links and broadcast ship the whole batch; + // shuffles slice it into one sub-batch per receiver. + def emitColumnarBatch(root: VectorSchemaRoot): Unit = { + val rowCount = root.getRowCount + val texeraSchema = ArrowUtils.toTexeraSchema(root.getSchema) + lazy val wholeBatchBytes = ArrowUtils.serializeRoot(root) + + partitioners.foreach { + case (_, partitioner) => + val receivers = partitioner.allReceivers + partitioner match { + case _ if receivers.size == 1 => + outputGateway.sendTo( + receivers.head, + ColumnarFrame(wholeBatchBytes, rowCount, texeraSchema) + ) + case _: BroadcastPartitioner => + receivers.foreach(r => + outputGateway.sendTo(r, ColumnarFrame(wholeBatchBytes, rowCount, texeraSchema)) + ) + case _ => + // Shuffle/round-robin: assign each row to a receiver via the + // partitioner, then ship one Arrow sub-batch per receiver. The + // row decode here is only to compute the partition key; downstream + // still receives columnar data. + val masks = Array.fill(receivers.size)(new Array[Boolean](rowCount)) + var i = 0 + while (i < rowCount) { + val t = ArrowUtils.getTexeraTuple(i, root) + partitioner.getBucketIndex(t).foreach(b => masks(b)(i) = true) + i += 1 + } + var b = 0 + while (b < receivers.size) { + val sub = ArrowUtils.selectRows(root, masks(b), emitAllocator()) + try { + if (sub.getRowCount > 0) { + outputGateway.sendTo( + receivers(b), + ColumnarFrame(ArrowUtils.serializeRoot(sub), sub.getRowCount, texeraSchema) + ) + } + } finally sub.close() + b += 1 + } + } + } + + // Materialize to result storage for the GUI result panel. Hand the whole + // Arrow batch to the writer thread(s), which decode off the DP thread. Reuse + // the bytes already serialized for the wire when available. + if (outputPortResultWriterThreads.nonEmpty && rowCount > 0) { + saveArrowBatchToStorageIfNeeded(wholeBatchBytes) + } + } + + // Columnar batches pending emit, drained one-per-step by the DP loop so that + // backpressure applies between batches (unlike a synchronous emit). + private var columnarOutputIter: Iterator[VectorSchemaRoot] = Iterator.empty + def setColumnarOutput(it: Iterator[VectorSchemaRoot]): Unit = columnarOutputIter = it + def hasUnfinishedColumnarOutput: Boolean = columnarOutputIter.hasNext + def emitOneColumnarBatch(): Unit = { + val root = columnarOutputIter.next() + emitColumnarBatch(root) + root.close() + } + def addPort(portId: PortIdentity, schema: Schema, storageURIBaseOption: Option[URI]): Unit = { // each port can only be added and initialized once. if (this.ports.contains(portId)) { @@ -239,10 +326,19 @@ class OutputManager( }).foreach({ case (portId, writerThread) => // write to storage in a separate thread - writerThread.queue.put(Left(tuple)) + writerThread.queue.put(Left(RowWriteItem(tuple))) }) } + // Columnar sink: hand the whole Arrow batch to each result writer thread, + // which decodes it to rows off the DP thread. Avoids the per-row decode on + // the hot path and reuses the batch bytes already produced for the wire. + def saveArrowBatchToStorageIfNeeded(arrowIpcBytes: Array[Byte]): Unit = { + outputPortResultWriterThreads.values.foreach( + _.queue.put(Left(ArrowBatchWriteItem(arrowIpcBytes))) + ) + } + private def saveStateToStorageIfNeeded( state: State, loopCounter: Long, @@ -254,7 +350,9 @@ class OutputManager( // downstream operator (and every worker reading the materialization) // needs the full set. The loop envelope is materialized as its own // columns so the downstream reader can rebuild it. - stateWriterThreads.values.foreach(_.queue.put(Left(state.toTuple(loopCounter, loopStartId)))) + stateWriterThreads.values.foreach( + _.queue.put(Left(RowWriteItem(state.toTuple(loopCounter, loopStartId)))) + ) } /** diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonProxyClient.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonProxyClient.scala index d3ea86cdd4a..6726138ff9f 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonProxyClient.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/pythonworker/PythonProxyClient.scala @@ -124,6 +124,13 @@ class PythonProxyClient(portNumberPromise: Promise[Int], val actorId: ActorVirtu dataPayload match { case DataFrame(frame) => writeArrowStream(mutable.Queue(ArraySeq.unsafeWrapArray(frame): _*), from, "Data") + case ColumnarFrame(bytes, _, _) => + // Columnar wire reaching the Python path: pass the Arrow batch straight + // to Flight (no tuple round-trip). deserializeRootFold keeps the root + // alive for the put and closes it after. + ArrowUtils.deserializeRootFold(bytes, allocator) { root => + writeArrowRoot(root, from, "Data") + } case StateFrame(state, loopCounter, loopStartId) => // The Arrow wire format for states IS the State row (content + // loop_counter + loop_start_id), so the envelope rides its own @@ -207,6 +214,24 @@ class PythonProxyClient(portNumberPromise: Promise[Int], val actorId: ActorVirtu result } + // Send an already-built Arrow batch to the Python worker over Flight, without + // decoding to tuples first. Used for the columnar wire passthrough. + private def writeArrowRoot( + root: VectorSchemaRoot, + from: ChannelIdentity, + payloadType: String + ): Unit = { + val descriptor = FlightDescriptor.command(PythonDataHeader(from, payloadType).toByteArray) + val flightListener = new SyncPutListener + val writer = flightClient.startPut(descriptor, root, flightListener) + writer.putNext() + writer.completed() + val ackMsgBuf: ArrowBuf = flightListener.poll(5, TimeUnit.SECONDS).getApplicationMetadata + pythonQueueInMemSize.set(ackMsgBuf.getLong(0)) + ackMsgBuf.close() + flightListener.close() + } + private def writeArrowStream( tuples: mutable.Queue[Tuple], from: ChannelIdentity, diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/Partitioner.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/Partitioner.scala index 620eb936512..8a2e4155d5f 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/Partitioner.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/sendsemantics/partitioners/Partitioner.scala @@ -24,7 +24,8 @@ import org.apache.texera.amber.core.state.State import org.apache.texera.amber.core.tuple.Tuple import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity import org.apache.texera.amber.engine.architecture.messaginglayer.NetworkOutputGateway -import org.apache.texera.amber.engine.common.ambermessage.{DataFrame, StateFrame} +import org.apache.texera.amber.engine.common.ambermessage.{ColumnarFrame, DataFrame, StateFrame} +import org.apache.texera.amber.util.ArrowUtils import scala.collection.mutable.ArrayBuffer @@ -57,9 +58,25 @@ class NetworkOutputBuffer( def flush(): Unit = { if (buffer.nonEmpty) { - dataOutputPort.sendTo(to, DataFrame(buffer.toArray)) + val batch = buffer.toArray + // Columnar wire (flagged): send the batch as an Arrow IPC ColumnarFrame. + val payload = + if (NetworkOutputBuffer.columnarWire) + ColumnarFrame( + ArrowUtils.serializeTuples(batch.head.getSchema, batch), + batch.length, + batch.head.getSchema + ) + else DataFrame(batch) + dataOutputPort.sendTo(to, payload) buffer = new ArrayBuffer[Tuple]() } } } + +object NetworkOutputBuffer { + // Opt-in for the Arrow columnar wire format (default off = row DataFrame), + // from application.conf (columnar.enable-columnar-wire), COLUMNAR_WIRE overrides. + val columnarWire: Boolean = ApplicationConfig.enableColumnarWire +} diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala index 829eaf1a77e..f199f2ffb35 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DPThread.scala @@ -165,7 +165,8 @@ class DPThread( var channelId: ChannelIdentity = null var msgOpt: Option[WorkflowFIFOMessage] = None if ( - dp.inputManager.hasUnfinishedInput || dp.outputManager.hasUnfinishedOutput || dp.pauseManager.isPaused + dp.inputManager.hasUnfinishedInput || dp.outputManager.hasUnfinishedOutput || + dp.outputManager.hasUnfinishedColumnarOutput || dp.pauseManager.isPaused ) { dp.inputGateway.tryPickControlChannel match { case Some(channel) => diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala index 618fa73427a..aa45c8471ba 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala @@ -21,7 +21,12 @@ package org.apache.texera.amber.engine.architecture.worker import com.softwaremill.macwire.wire import io.grpc.MethodDescriptor -import org.apache.texera.amber.core.executor.OperatorExecutor +import org.apache.texera.amber.core.executor.{ + ColumnarOperatorExecutor, + ColumnarResult, + OperatorExecutor +} +import org.apache.texera.amber.engine.architecture.sendsemantics.partitioners.NetworkOutputBuffer import org.apache.texera.amber.core.state.State import org.apache.texera.amber.core.tuple._ import org.apache.texera.amber.core.virtualidentity.{ @@ -59,6 +64,7 @@ import org.apache.texera.amber.engine.common.ambermessage._ import org.apache.texera.amber.engine.common.statetransition.WorkerStateManager import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR import org.apache.texera.amber.error.ErrorUtils.{mkConsoleMessage, safely} +import org.apache.texera.amber.util.ArrowUtils import java.util.concurrent.LinkedBlockingQueue @@ -169,6 +175,7 @@ class DataProcessor( sendECMToDataChannels(METHOD_END_CHANNEL, PORT_ALIGNMENT) // Send Completed signal to worker actor. executor.close() + outputManager.closeColumnarResources() adaptiveBatchingMonitor.stopAdaptiveBatching() stateManager.transitTo(COMPLETED) logger.debug( @@ -201,7 +208,9 @@ class DataProcessor( def continueDataProcessing(): Unit = { val dataProcessingStartTime = System.nanoTime() - if (outputManager.hasUnfinishedOutput) { + if (outputManager.hasUnfinishedColumnarOutput) { + outputManager.emitOneColumnarBatch() + } else if (outputManager.hasUnfinishedOutput) { outputOneTuple() } else { processInputTuple(inputManager.getNextTuple) @@ -217,25 +226,74 @@ class DataProcessor( val portId = this.inputGateway.getChannel(channelId).getPortId dataPayload match { case DataFrame(tuples) => - stateManager.conditionalTransitTo( - READY, - RUNNING, - () => { - val (state, stateVersion) = stateManager.getStateWithVersion - asyncRPCClient.coordinatorInterface.workerStateUpdated( - WorkerStateUpdatedRequest(state, stateVersion), - asyncRPCClient.mkContext(COORDINATOR) - ) - } - ) - inputManager.initBatch(channelId, tuples) - processInputTuple(inputManager.getNextTuple) + processTupleBatch(channelId, portId, tuples) + case ColumnarFrame(bytes, _, _) => + executor match { + // Native-Arrow path: consume the batch directly, emit a filtered batch. + case c: ColumnarOperatorExecutor if NetworkOutputBuffer.columnarWire => + c.processColumnarBatch(bytes, portId.id) match { + case ColumnarResult.Emit(result) => + logColumnarModeOnce(active = true, "") + statisticsManager.increaseInputStatistics(portId, bytes.length.toLong) + outputManager.setColumnarOutput(Iterator.single(result)) + case ColumnarResult.EmitRows(rows) => + logColumnarModeOnce(active = true, "") + statisticsManager.increaseInputStatistics(portId, bytes.length.toLong) + outputManager.outputIterator.setTupleOutput(rows) + case ColumnarResult.Consumed => + logColumnarModeOnce(active = true, "") + statisticsManager.increaseInputStatistics(portId, bytes.length.toLong) + case ColumnarResult.Unsupported => + logColumnarModeOnce( + active = false, + "operator has no native columnar path for this batch" + ) + processTupleBatch(channelId, portId, ArrowUtils.deserializeTuples(bytes)) + } + case _ => + processTupleBatch(channelId, portId, ArrowUtils.deserializeTuples(bytes)) + } case StateFrame(state, loopCounter, loopStartId) => processInputState(state, portId.id, loopCounter, loopStartId) } statisticsManager.increaseDataProcessingTime(System.nanoTime() - dataProcessingStartTime) } + private def processTupleBatch( + channelId: ChannelIdentity, + portId: PortIdentity, + tuples: Array[Tuple] + ): Unit = { + stateManager.conditionalTransitTo( + READY, + RUNNING, + () => { + val (state, stateVersion) = stateManager.getStateWithVersion + asyncRPCClient.coordinatorInterface.workerStateUpdated( + WorkerStateUpdatedRequest(state, stateVersion), + asyncRPCClient.mkContext(COORDINATOR) + ) + } + ) + inputManager.initBatch(channelId, tuples) + // Whole-batch path if the executor supports it, else per-tuple; errors fall back. + val vectorized: Option[Iterator[(TupleLike, Option[PortIdentity])]] = + try executor.processBatchMultiPort(tuples, portId.id) + catch safely { case _ => None } + vectorized match { + case Some(outputIter) => + var i = 0 + while (i < tuples.length) { + statisticsManager.increaseInputStatistics(portId, tuples(i).inMemSize) + i += 1 + } + inputManager.skipToEnd() + outputManager.outputIterator.setTupleOutput(outputIter) + case None => + processInputTuple(inputManager.getNextTuple) + } + } + def processECM( channelId: ChannelIdentity, ecm: EmbeddedControlMessage, @@ -309,6 +367,17 @@ class DataProcessor( } } + // Log the columnar-wire decision once per worker so a silent fall back to the + // row path is visible in the logs. + @transient private var columnarModeLogged = false + private[architecture] def logColumnarModeOnce(active: Boolean, reason: String): Unit = { + if (!columnarModeLogged) { + columnarModeLogged = true + if (active) logger.info(s"columnar wire active for $executor") + else logger.info(s"columnar wire requested but using row path for $executor: $reason") + } + } + def handleExecutorException(e: Throwable): Unit = { asyncRPCClient.coordinatorInterface.consoleMessageTriggered( ConsoleMessageTriggeredRequest(mkConsoleMessage(actorId, e)), diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThread.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThread.scala index f0b814da1de..5495aa4d8ce 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThread.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThread.scala @@ -22,6 +22,7 @@ package org.apache.texera.amber.engine.architecture.worker.managers import com.google.common.collect.Queues import org.apache.texera.amber.core.storage.model.BufferedItemWriter import org.apache.texera.amber.core.tuple.Tuple +import org.apache.texera.amber.util.ArrowUtils import java.util.concurrent.LinkedBlockingQueue import scala.util.control.NonFatal @@ -29,12 +30,18 @@ import scala.util.control.NonFatal sealed trait TerminateSignal case object PortStorageWriterTerminateSignal extends TerminateSignal +// A unit of work for the writer thread: a single row, or a whole Arrow batch +// (columnar sink) decoded on this thread instead of on the hot DP thread. +sealed trait StorageWriteItem +final case class RowWriteItem(tuple: Tuple) extends StorageWriteItem +final case class ArrowBatchWriteItem(arrowIpcBytes: Array[Byte]) extends StorageWriteItem + class OutputPortStorageWriterThread( bufferedItemWriter: BufferedItemWriter[Tuple] ) extends Thread { - val queue: LinkedBlockingQueue[Either[Tuple, TerminateSignal]] = - Queues.newLinkedBlockingQueue[Either[Tuple, TerminateSignal]]() + val queue: LinkedBlockingQueue[Either[StorageWriteItem, TerminateSignal]] = + Queues.newLinkedBlockingQueue[Either[StorageWriteItem, TerminateSignal]]() // Captured failure from put-one or close() so the worker DP thread can // re-throw and let the coordinator's pekko supervisor surface a FatalError @@ -50,8 +57,11 @@ class OutputPortStorageWriterThread( var internalStop = false while (!internalStop) { queue.take() match { - case Left(tuple) => bufferedItemWriter.putOne(tuple) - case Right(_) => internalStop = true + case Left(RowWriteItem(tuple)) => bufferedItemWriter.putOne(tuple) + case Left(ArrowBatchWriteItem(bytes)) => + // Decode the Arrow batch here (off the DP thread) into rows. + ArrowUtils.deserializeTuples(bytes).foreach(bufferedItemWriter.putOne) + case Right(_) => internalStop = true } } } catch { diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala index f1b37d7d325..8da4de9fab5 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/promisehandlers/EndChannelHandler.scala @@ -20,12 +20,14 @@ package org.apache.texera.amber.engine.architecture.worker.promisehandlers import com.twitter.util.Future +import org.apache.texera.amber.core.executor.SourceOperatorExecutor import org.apache.texera.amber.core.tuple.FinalizePort import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ AsyncRPCContext, EmptyRequest } import org.apache.texera.amber.engine.architecture.rpc.controlreturns.EmptyReturn +import org.apache.texera.amber.engine.architecture.sendsemantics.partitioners.NetworkOutputBuffer import org.apache.texera.amber.engine.architecture.worker.DataProcessorRPCHandlerInitializer import org.apache.texera.amber.error.ErrorUtils.safely @@ -48,9 +50,32 @@ trait EndChannelHandler { // `main_loop._process_state_frame` for how a Loop End treats it. dp.outputManager.emitState(outputState.get) } - dp.outputManager.outputIterator.setTupleOutput( - dp.executor.onFinishMultiPort(portId.id) - ) + // Columnar source path: emit Arrow batches directly (no per-row Tuple), + // when enabled and there are output links (shuffles are split in emit). + // Else the row-oriented onFinishMultiPort path. + val columnarBatches = + if (NetworkOutputBuffer.columnarWire) + dp.executor match { + case s: SourceOperatorExecutor => s.produceColumnarBatch() + case _ => None + } + else None + columnarBatches match { + case Some(batchIter) => + dp.logColumnarModeOnce(active = true, "") + // Drained one batch per DP-loop step (backpressure applies between). + dp.outputManager.setColumnarOutput(batchIter) + dp.outputManager.outputIterator.setTupleOutput(Iterator.empty) + case None => + if ( + NetworkOutputBuffer.columnarWire && dp.executor.isInstanceOf[SourceOperatorExecutor] + ) { + dp.logColumnarModeOnce(active = false, "source has no native columnar batch producer") + } + dp.outputManager.outputIterator.setTupleOutput( + dp.executor.onFinishMultiPort(portId.id) + ) + } } catch safely { case e => // forward input tuple to the user and pause DP thread diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/DataPayload.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/DataPayload.scala index b8ac640ad53..cd8db757db3 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/DataPayload.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/DataPayload.scala @@ -20,7 +20,7 @@ package org.apache.texera.amber.engine.common.ambermessage import org.apache.texera.amber.core.state.State -import org.apache.texera.amber.core.tuple.Tuple +import org.apache.texera.amber.core.tuple.{Schema, Tuple} sealed trait DataPayload extends WorkflowFIFOMessagePayload {} @@ -37,6 +37,12 @@ sealed trait DataPayload extends WorkflowFIFOMessagePayload {} final case class StateFrame(frame: State, loopCounter: Long = 0L, loopStartId: String = "") extends DataPayload +// Columnar wire payload: a batch encoded as Arrow IPC stream bytes. +final case class ColumnarFrame(arrowIpcBytes: Array[Byte], rowCount: Int, schema: Schema) + extends DataPayload { + val inMemSize: Long = arrowIpcBytes.length.toLong +} + final case class DataFrame(frame: Array[Tuple]) extends DataPayload { val inMemSize: Long = { frame.map(_.inMemSize).sum diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/WorkflowMessage.scala b/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/WorkflowMessage.scala index a54d4c20310..e67878476ab 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/WorkflowMessage.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/common/ambermessage/WorkflowMessage.scala @@ -26,8 +26,9 @@ case object WorkflowMessage { msg match { case dataMsg: WorkflowFIFOMessage => dataMsg.payload match { - case df: DataFrame => df.inMemSize - case _ => 200L + case df: DataFrame => df.inMemSize + case cf: ColumnarFrame => cf.inMemSize + case _ => 200L } case _ => 200L } diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThreadSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThreadSpec.scala index 9cc8216efab..50b922b8094 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThreadSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/managers/OutputPortStorageWriterThreadSpec.scala @@ -75,7 +75,7 @@ class OutputPortStorageWriterThreadSpec extends AnyFlatSpec { val writer = new StubWriter(onPutOne = throwing("test putOne failure")) val thread = new OutputPortStorageWriterThread(writer) thread.start() - thread.queue.put(Left(null.asInstanceOf[Tuple])) + thread.queue.put(Left(RowWriteItem(null.asInstanceOf[Tuple]))) thread.queue.put(Right(PortStorageWriterTerminateSignal)) thread.join() assert(thread.getFailure.exists(_.getMessage.contains("test putOne failure"))) @@ -91,7 +91,7 @@ class OutputPortStorageWriterThreadSpec extends AnyFlatSpec { ) val thread = new OutputPortStorageWriterThread(writer) thread.start() - thread.queue.put(Left(null.asInstanceOf[Tuple])) + thread.queue.put(Left(RowWriteItem(null.asInstanceOf[Tuple]))) thread.queue.put(Right(PortStorageWriterTerminateSignal)) thread.join() val captured = thread.getFailure.getOrElse(fail("expected putOne failure")) diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ColumnarReproSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ColumnarReproSpec.scala new file mode 100644 index 00000000000..2a9e3dc0a71 --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ColumnarReproSpec.scala @@ -0,0 +1,126 @@ +/* + * 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.texera.amber.engine.e2e + +import org.apache.pekko.actor.{ActorSystem, Props} +import org.apache.pekko.testkit.{ImplicitSender, TestKit} +import org.apache.pekko.util.Timeout +import org.apache.texera.amber.clustering.SingleNodeListener +import org.apache.texera.amber.core.tuple.Tuple +import org.apache.texera.amber.core.virtualidentity.OperatorIdentity +import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} +import org.apache.texera.amber.engine.common.AmberRuntime +import org.apache.texera.amber.engine.e2e.TestUtils.{ + buildWorkflow, + cleanupWorkflowExecutionData, + initiateTexeraDBForTestCases, + runWorkflowAndReadTerminalResults, + setUpWorkflowExecutionData +} +import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.filter.{ + ComparisonType, + FilterPredicate, + SpecializedFilterOpDesc +} +import org.apache.texera.common.compiler.model.LogicalLink +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.BeforeAndAfterAll + +import com.twitter.util.Duration +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.duration.DurationInt + +class ColumnarReproSpec + extends TestKit(ActorSystem("ColumnarReproSpec", AmberRuntime.pekkoConfig)) + with ImplicitSender + with AnyFlatSpecLike + with BeforeAndAfterAll { + + implicit val timeout: Timeout = Timeout(5.seconds) + private val ids = new AtomicInteger(70000) + private val tag = if (sys.env.get("COLUMNAR_WIRE").contains("1")) "COLUMNAR" else "ROW" + + override def beforeAll(): Unit = { + system.actorOf(Props[SingleNodeListener](), "cluster-info") + Class.forName("org.postgresql.Driver") + initiateTexeraDBForTestCases() + } + override def afterAll(): Unit = TestKit.shutdownActorSystem(system) + + private def filt(attr: String, v: String): SpecializedFilterOpDesc = { + val op = new SpecializedFilterOpDesc() + op.predicates = List(new FilterPredicate(attr, ComparisonType.GREATER_THAN, v)); op + } + + private def report(name: String, res: Map[OperatorIdentity, List[Tuple]]): Unit = { + val rows = res.values.headOption.getOrElse(Nil) + println(s"REPRO[$tag] $name: ${rows.size} rows") + rows.take(3).foreach { t => + println( + s"REPRO[$tag] Region=${t.getField[Any]("Region")} UnitsSold=${t.getField[Any]("Units Sold")}" + ) + } + } + + "columnar source" should "scan -> filter(terminal)" in { + val id = ids.incrementAndGet(); setUpWorkflowExecutionData(id) + try { + val ctx: WorkflowContext = TestUtils.workflowContext(id) + val scan = TestOperators.smallCsvScanOpDesc() + val f = filt("Units Sold", "5000") + val wf = buildWorkflow( + List(scan, f), + List( + LogicalLink(scan.operatorIdentifier, PortIdentity(), f.operatorIdentifier, PortIdentity()) + ), + ctx + ) + report("scan->filter", runWorkflowAndReadTerminalResults(system, wf, Duration.fromMinutes(5))) + } finally cleanupWorkflowExecutionData(id) + } + + "native filter" should "scan -> filter -> filter(terminal)" in { + val id = ids.incrementAndGet(); setUpWorkflowExecutionData(id) + try { + val ctx: WorkflowContext = TestUtils.workflowContext(id) + val scan = TestOperators.smallCsvScanOpDesc() + val f1 = filt("Units Sold", "5000") + val f2 = filt("Units Sold", "0") + val wf = buildWorkflow( + List(scan, f1, f2), + List( + LogicalLink( + scan.operatorIdentifier, + PortIdentity(), + f1.operatorIdentifier, + PortIdentity() + ), + LogicalLink(f1.operatorIdentifier, PortIdentity(), f2.operatorIdentifier, PortIdentity()) + ), + ctx + ) + report( + "scan->filter->filter", + runWorkflowAndReadTerminalResults(system, wf, Duration.fromMinutes(5)) + ) + } finally cleanupWorkflowExecutionData(id) + } +} diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ColumnarShuffleCorrectnessSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ColumnarShuffleCorrectnessSpec.scala new file mode 100644 index 00000000000..e8f258b1009 --- /dev/null +++ b/amber/src/test/scala/org/apache/texera/amber/engine/e2e/ColumnarShuffleCorrectnessSpec.scala @@ -0,0 +1,246 @@ +/* + * 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.texera.amber.engine.e2e + +import org.apache.pekko.actor.{ActorSystem, Props} +import org.apache.pekko.testkit.{ImplicitSender, TestKit} +import org.apache.pekko.util.Timeout +import org.apache.texera.amber.clustering.SingleNodeListener +import org.apache.texera.amber.core.tuple.Tuple +import org.apache.texera.amber.core.virtualidentity.OperatorIdentity +import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} +import org.apache.texera.amber.engine.common.AmberRuntime +import org.apache.texera.amber.engine.e2e.TestUtils.{ + buildWorkflow, + cleanupWorkflowExecutionData, + initiateTexeraDBForTestCases, + runWorkflowAndReadTerminalResults, + setUpWorkflowExecutionData +} +import org.apache.texera.amber.operator.TestOperators +import org.apache.texera.amber.operator.aggregate.{ + AggregateOpDesc, + AggregationFunction, + AggregationOperation +} +import org.apache.texera.amber.operator.filter.{ + ComparisonType, + FilterPredicate, + SpecializedFilterOpDesc +} +import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} +import org.apache.texera.common.compiler.model.LogicalLink +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpecLike + +import com.twitter.util.Duration +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.duration.DurationInt + +/** + * Multi-worker columnar correctness: scan -> filter -> count-by-group. The + * aggregate's group-by induces a hash-shuffle edge upstream, so with more than + * one worker the columnar path must split each Arrow batch per receiver. Run + * this under COLUMNAR_WIRE=0 and =1 (with CONSTANTS_NUM_WORKER_PER_OPERATOR=2) + * and diff the SHUFFLE lines: the group counts must match. Proves the columnar + * shuffle partitioner produces the same result as the row path. + */ +class ColumnarShuffleCorrectnessSpec + extends TestKit(ActorSystem("ColumnarShuffleCorrectnessSpec", AmberRuntime.pekkoConfig)) + with ImplicitSender + with AnyFlatSpecLike + with BeforeAndAfterAll { + + implicit val timeout: Timeout = Timeout(5.seconds) + private val ids = new AtomicInteger(90000) + private val tag = if (sys.env.get("COLUMNAR_WIRE").contains("1")) "COLUMNAR" else "ROW" + private val workers = sys.env.getOrElse("CONSTANTS_NUM_WORKER_PER_OPERATOR", "1") + + override def beforeAll(): Unit = { + system.actorOf(Props[SingleNodeListener](), "cluster-info") + Class.forName("org.postgresql.Driver") + initiateTexeraDBForTestCases() + } + override def afterAll(): Unit = TestKit.shutdownActorSystem(system) + + private def filterGT(attr: String, v: String): SpecializedFilterOpDesc = { + val op = new SpecializedFilterOpDesc() + op.predicates = List(new FilterPredicate(attr, ComparisonType.GREATER_THAN, v)); op + } + private def agg(fn: AggregationFunction, attr: String, res: String): AggregationOperation = { + val a = new AggregationOperation(); a.aggFunction = fn; a.attribute = attr; + a.resultAttribute = res; a + } + private def aggByRegion(): AggregateOpDesc = { + val op = new AggregateOpDesc() + op.aggregations = List( + agg(AggregationFunction.COUNT, "Region", "cnt"), + agg(AggregationFunction.SUM, "Units Sold", "sum_units"), + agg(AggregationFunction.MIN, "Units Sold", "min_units"), + agg(AggregationFunction.MAX, "Units Sold", "max_units"), + agg(AggregationFunction.AVERAGE, "Units Sold", "avg_units") + ) + op.groupByKeys = List("Region"); op + } + + private def report(res: Map[OperatorIdentity, List[Tuple]]): Unit = { + val rows = res.values.headOption.getOrElse(Nil) + val lines = rows + .map(t => + s"${t.getField[Any]("Region")} cnt=${t.getField[Any]("cnt")} sum=${t.getField[Any]("sum_units")} " + + s"min=${t.getField[Any]("min_units")} max=${t.getField[Any]("max_units")} avg=${t + .getField[Any]("avg_units")}" + ) + .sorted + println(s"SHUFFLE[$tag workers=$workers] groups=${rows.size}") + lines.foreach(l => println(s"SHUFFLE[$tag] $l")) + } + + "columnar shuffle" should "scan -> filter -> agg-by-Region match the row path" in { + val id = ids.incrementAndGet(); setUpWorkflowExecutionData(id) + try { + val ctx: WorkflowContext = TestUtils.workflowContext(id) + val scan = TestOperators.smallCsvScanOpDesc() + val f = filterGT("Units Sold", "5000") + val aggOp = aggByRegion() + val wf = buildWorkflow( + List(scan, f, aggOp), + List( + LogicalLink( + scan.operatorIdentifier, + PortIdentity(), + f.operatorIdentifier, + PortIdentity() + ), + LogicalLink( + f.operatorIdentifier, + PortIdentity(), + aggOp.operatorIdentifier, + PortIdentity() + ) + ), + ctx + ) + report(runWorkflowAndReadTerminalResults(system, wf, Duration.fromMinutes(5))) + } finally cleanupWorkflowExecutionData(id) + } + + private def projectRegionUnits(): ProjectionOpDesc = { + val op = new ProjectionOpDesc() + op.attributes = + List(new AttributeUnit("Region", "Region"), new AttributeUnit("Units Sold", "units")) + op + } + private def countSumByRegionOn(unitsCol: String): AggregateOpDesc = { + val op = new AggregateOpDesc() + op.aggregations = List( + agg(AggregationFunction.COUNT, "Region", "cnt"), + agg(AggregationFunction.SUM, unitsCol, "sum_units") + ) + op.groupByKeys = List("Region"); op + } + private def reportCntSum(res: Map[OperatorIdentity, List[Tuple]]): Unit = { + val rows = res.values.headOption.getOrElse(Nil) + val lines = + rows + .map(t => + s"${t.getField[Any]("Region")} cnt=${t.getField[Any]("cnt")} sum=${t.getField[Any]("sum_units")}" + ) + .sorted + println(s"PROJ[$tag workers=$workers] groups=${rows.size}") + lines.foreach(l => println(s"PROJ[$tag] $l")) + } + + private def reportJoin(res: Map[OperatorIdentity, List[Tuple]]): Unit = { + val rows = res.values.headOption.getOrElse(Nil) + val checksum = rows.map(_.getFields.mkString("|")).sorted.mkString("\n").hashCode + println(s"JOIN[$tag workers=$workers] rows=${rows.size} checksum=$checksum") + } + + // Selective columnar probe: the probe decodes only the join key per row and + // fully decodes a row only on a match. Row count + content checksum must match + // the row path. + "columnar join" should "csv join csv on column-1 match the row path" in { + val id = ids.incrementAndGet(); setUpWorkflowExecutionData(id) + try { + val ctx: WorkflowContext = TestUtils.workflowContext(id) + val c1 = TestOperators.headerlessSmallCsvScanOpDesc() + val c2 = TestOperators.headerlessSmallCsvScanOpDesc() + val join = TestOperators.joinOpDesc("column-1", "column-1") + val wf = buildWorkflow( + List(c1, c2, join), + List( + LogicalLink( + c1.operatorIdentifier, + PortIdentity(), + join.operatorIdentifier, + PortIdentity() + ), + LogicalLink( + c2.operatorIdentifier, + PortIdentity(), + join.operatorIdentifier, + PortIdentity(1) + ) + ), + ctx + ) + reportJoin(runWorkflowAndReadTerminalResults(system, wf, Duration.fromMinutes(5))) + } finally cleanupWorkflowExecutionData(id) + } + + // Projection renames "Units Sold" -> "units"; the downstream filter and + // aggregate then reference the renamed column, all over the columnar wire. + "columnar projection" should "scan -> project(rename) -> filter -> agg match the row path" in { + val id = ids.incrementAndGet(); setUpWorkflowExecutionData(id) + try { + val ctx: WorkflowContext = TestUtils.workflowContext(id) + val scan = TestOperators.smallCsvScanOpDesc() + val proj = projectRegionUnits() + val f = filterGT("units", "5000") + val aggOp = countSumByRegionOn("units") + val wf = buildWorkflow( + List(scan, proj, f, aggOp), + List( + LogicalLink( + scan.operatorIdentifier, + PortIdentity(), + proj.operatorIdentifier, + PortIdentity() + ), + LogicalLink( + proj.operatorIdentifier, + PortIdentity(), + f.operatorIdentifier, + PortIdentity() + ), + LogicalLink( + f.operatorIdentifier, + PortIdentity(), + aggOp.operatorIdentifier, + PortIdentity() + ) + ), + ctx + ) + reportCntSum(runWorkflowAndReadTerminalResults(system, wf, Duration.fromMinutes(5))) + } finally cleanupWorkflowExecutionData(id) + } +} diff --git a/common/config/src/main/resources/application.conf b/common/config/src/main/resources/application.conf index d5f7ffed2e7..6cd9bc3f38e 100644 --- a/common/config/src/main/resources/application.conf +++ b/common/config/src/main/resources/application.conf @@ -60,6 +60,15 @@ reconfiguration { enable-transactional-reconfiguration = ${?RECONFIGURATION_ENABLE_TRANSACTIONAL_RECONFIGURATION} } +columnar { + # Arrow columnar wire format between operators (opt-in). The COLUMNAR_WIRE + # env var ("1"/"0") still overrides this, applied in ApplicationConfig. + enable-columnar-wire = false + + # Vectorized path in operators that support it (e.g. the filter). + enable-vectorized-operators = true +} + cache { # [false, true] enabled = true diff --git a/common/config/src/main/scala/org/apache/texera/common/config/ApplicationConfig.scala b/common/config/src/main/scala/org/apache/texera/common/config/ApplicationConfig.scala index dea1e169d6a..017b020abf4 100644 --- a/common/config/src/main/scala/org/apache/texera/common/config/ApplicationConfig.scala +++ b/common/config/src/main/scala/org/apache/texera/common/config/ApplicationConfig.scala @@ -71,6 +71,19 @@ object ApplicationConfig { val enableTransactionalReconfiguration: Boolean = getConfSource.getBoolean("reconfiguration.enable-transactional-reconfiguration") + // Columnar execution. The COLUMNAR_WIRE env ("1"/"0") is honored for backward + // compatibility and overrides the config value when set. + val enableColumnarWire: Boolean = + sys.env.get("COLUMNAR_WIRE") match { + case Some(v) => v == "1" || v.equalsIgnoreCase("true") + case None => getConfSource.getBoolean("columnar.enable-columnar-wire") + } + val enableVectorizedOperators: Boolean = + sys.env.get("FILTER_VECTORIZED") match { + case Some(v) => v == "1" || v.equalsIgnoreCase("true") + case None => getConfSource.getBoolean("columnar.enable-vectorized-operators") + } + // Fault tolerance val faultToleranceLogFlushIntervalInMs: Long = getConfSource.getLong("fault-tolerance.log-flush-interval-ms") diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/ColumnarOperatorExecutor.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/ColumnarOperatorExecutor.scala new file mode 100644 index 00000000000..853004b32e1 --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/ColumnarOperatorExecutor.scala @@ -0,0 +1,52 @@ +/* + * 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.texera.amber.core.executor + +import org.apache.arrow.vector.VectorSchemaRoot +import org.apache.texera.amber.core.tuple.TupleLike +import org.apache.texera.amber.core.workflow.PortIdentity + +/** + * Outcome of consuming one Arrow batch: + * - Emit: consumed, emit this Arrow result batch downstream (caller serializes and closes the root). + * - EmitRows: consumed, emit these row outputs (e.g. a join whose output is row-shaped); + * the caller ships them, still columnar on the wire if enabled. + * - Consumed: consumed, nothing to emit now (a blocking operator accumulating state). + * - Unsupported: not handled, caller falls back to the row path (decode + processTuple). + */ +sealed trait ColumnarResult +object ColumnarResult { + final case class Emit(root: VectorSchemaRoot) extends ColumnarResult + final case class EmitRows(rows: Iterator[(TupleLike, Option[PortIdentity])]) + extends ColumnarResult + case object Consumed extends ColumnarResult + case object Unsupported extends ColumnarResult +} + +/** + * An operator that can consume an Arrow columnar batch directly (no per-row + * Tuple decode). Given the incoming batch as Arrow IPC bytes and the input + * port it arrived on, returns a ColumnarResult telling the caller whether it + * emitted a batch/rows, consumed the batch with no output, or could not handle + * it (fall back to the row path). + */ +trait ColumnarOperatorExecutor { + def processColumnarBatch(arrowIpcBytes: Array[Byte], port: Int): ColumnarResult +} diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala index 9837213abbb..1440c5eedd2 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala @@ -40,6 +40,12 @@ trait OperatorExecutor { def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] + /** Optional whole-batch processing; None means process per-tuple via processTuple. */ + def processBatchMultiPort( + batch: Array[Tuple], + port: Int + ): Option[Iterator[(TupleLike, Option[PortIdentity])]] = None + def produceStateOnFinish(port: Int): Option[State] = None def onFinishMultiPort(port: Int): Iterator[(TupleLike, Option[PortIdentity])] = { diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/SourceOperatorExecutor.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/SourceOperatorExecutor.scala index 633314679a4..24c21b9997c 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/SourceOperatorExecutor.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/SourceOperatorExecutor.scala @@ -19,12 +19,20 @@ package org.apache.texera.amber.core.executor +import org.apache.arrow.vector.VectorSchemaRoot import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} import org.apache.texera.amber.core.workflow.PortIdentity trait SourceOperatorExecutor extends OperatorExecutor { override def open(): Unit = {} + /** + * Optional columnar production: yields Arrow batches directly (no per-row + * Tuple objects). None means the engine uses the row-oriented produceTuple. + * The caller serializes and closes each returned root. + */ + def produceColumnarBatch(): Option[Iterator[VectorSchemaRoot]] = None + override def close(): Unit = {} override def processTupleMultiPort( tuple: Tuple, diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/tuple/ColumnarBatch.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/tuple/ColumnarBatch.scala new file mode 100644 index 00000000000..2161b16c8ef --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/tuple/ColumnarBatch.scala @@ -0,0 +1,76 @@ +/* + * 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.texera.amber.core.tuple + +/** + * Column-oriented view of a batch of tuples, for vectorized operator execution. + * Wraps the original rows and extracts one column at a time into a contiguous + * array, caching each extraction. A numeric column can be pulled into a + * primitive `Array[Double]` plus a null mask for tight, JIT/SIMD-friendly loops. + */ +class ColumnarBatch(val rows: Array[Tuple]) { + + def size: Int = rows.length + + def schema: Schema = if (rows.isEmpty) null else rows(0).getSchema + + private val cache = scala.collection.mutable.HashMap[String, Array[Any]]() + + /** Column values in row order (boxed), extracted once and cached. */ + def column(name: String): Array[Any] = { + cache.getOrElseUpdate( + name, { + val a = new Array[Any](rows.length) + var i = 0 + while (i < rows.length) { a(i) = rows(i).getField[Any](name); i += 1 } + a + } + ) + } + + /** + * Numeric column as primitive doubles plus a per-row null mask. NaN is stored + * where the value is null; callers must consult the mask. Only valid when the + * column type is INTEGER or DOUBLE. + */ + def doubleColumn(name: String): (Array[Double], Array[Boolean]) = { + val vals = new Array[Double](rows.length) + val isNull = new Array[Boolean](rows.length) + var i = 0 + while (i < rows.length) { + val f = rows(i).getField[Any](name) + if (f == null) { isNull(i) = true } + else vals(i) = f.asInstanceOf[Number].doubleValue() + i += 1 + } + (vals, isNull) + } + + /** The rows selected by `mask` (mask.length must equal size), as an iterator. */ + def select(mask: Array[Boolean]): Iterator[Tuple] = { + val out = scala.collection.mutable.ArrayBuffer[Tuple]() + var i = 0 + while (i < rows.length) { + if (mask(i)) out += rows(i) + i += 1 + } + out.iterator + } +} diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala index 434519b3dea..a797fbc8999 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/util/ArrowUtils.scala @@ -38,12 +38,16 @@ import org.apache.arrow.vector.{ VarCharVector, VectorSchemaRoot } +import org.apache.arrow.vector.ipc.{ArrowStreamReader, ArrowStreamWriter} +import java.io.{ByteArrayInputStream, ByteArrayOutputStream} +import java.nio.channels.Channels import java.nio.charset.StandardCharsets import java.sql.Timestamp import java.time.temporal.ChronoUnit import java.time.{Instant, LocalDateTime, ZoneId, ZoneOffset} import java.util +import scala.collection.mutable.ArrayBuffer import scala.jdk.CollectionConverters.CollectionHasAsScala import scala.language.implicitConversions @@ -55,6 +59,103 @@ object ArrowUtils extends LazyLogging { implicit def bool2int(b: Boolean): Int = if (b) 1 else 0 + // Serialize a VectorSchemaRoot to Arrow IPC stream bytes. Does not close root. + def serializeRoot(root: VectorSchemaRoot): Array[Byte] = { + val out = new ByteArrayOutputStream() + val writer = new ArrowStreamWriter(root, null, Channels.newChannel(out)) + try { writer.start(); writer.writeBatch(); writer.end() } + finally writer.close() + out.toByteArray + } + + // Serialize a batch of tuples to Arrow IPC stream bytes (columnar wire format). + def serializeTuples(schema: Schema, tuples: Array[Tuple]): Array[Byte] = { + val root = VectorSchemaRoot.create(fromTexeraSchema(schema), allocator) + try { + var i = 0 + while (i < tuples.length) { setTexeraTuple(tuples(i), i, root); i += 1 } + root.setRowCount(tuples.length) + serializeRoot(root) + } finally root.close() + } + + // Read one Arrow IPC batch, apply `f` to the live root, then close the reader. + // `f` must copy anything it needs out (the root is freed on return). + def deserializeRootFold[T](bytes: Array[Byte], allocator: BufferAllocator)( + f: VectorSchemaRoot => T + ): T = { + val reader = new ArrowStreamReader(new ByteArrayInputStream(bytes), allocator) + try { + reader.loadNextBatch() + f(reader.getVectorSchemaRoot) + } finally reader.close() + } + + // Copy the rows of `src` where keep(i) is true into a new independent root. + def selectRows( + src: VectorSchemaRoot, + keep: Array[Boolean], + allocator: BufferAllocator + ): VectorSchemaRoot = { + val dest = VectorSchemaRoot.create(src.getSchema, allocator) + val srcVecs = src.getFieldVectors + val destVecs = dest.getFieldVectors + var destIdx = 0 + var i = 0 + val n = src.getRowCount + while (i < n) { + if (keep(i)) { + var c = 0 + while (c < srcVecs.size) { + destVecs.get(c).copyFromSafe(i, destIdx, srcVecs.get(c)); c += 1 + } + destIdx += 1 + } + i += 1 + } + dest.setRowCount(destIdx) + dest + } + + // Build a new root keeping only the source columns at `srcIndices`, renamed to + // `outSchema`'s attribute names (columnar projection). Copies all rows. + def selectColumns( + src: VectorSchemaRoot, + srcIndices: Array[Int], + outSchema: Schema, + allocator: BufferAllocator + ): VectorSchemaRoot = { + val dest = VectorSchemaRoot.create(fromTexeraSchema(outSchema), allocator) + val n = src.getRowCount + val srcVecs = src.getFieldVectors + val destVecs = dest.getFieldVectors + var c = 0 + while (c < srcIndices.length) { + val sv = srcVecs.get(srcIndices(c)) + val dv = destVecs.get(c) + var i = 0 + while (i < n) { dv.copyFromSafe(i, i, sv); i += 1 } + c += 1 + } + dest.setRowCount(n) + dest + } + + // Inverse of serializeTuples: decode Arrow IPC stream bytes back to tuples. + def deserializeTuples(bytes: Array[Byte]): Array[Tuple] = { + val reader = new ArrowStreamReader(new ByteArrayInputStream(bytes), allocator) + try { + val out = ArrayBuffer[Tuple]() + val root = reader.getVectorSchemaRoot + while (reader.loadNextBatch()) { + val n = root.getRowCount + var i = 0 + while (i < n) { out += getTexeraTuple(i, root); i += 1 } + } + out.toArray + } finally reader.close() + } + /** * Reads a row of the given Arrow Vectors into a Texera.Tuple * e.g., @@ -101,6 +202,41 @@ object ArrowUtils extends LazyLogging { .build() } + // Source-column indices in `root` for the given attribute names, in order. + // Throws if a name is missing. Compute once per batch, reuse across rows. + def projectionIndices(root: VectorSchemaRoot, names: Seq[String]): Array[Int] = { + val fields = root.getSchema.getFields + names.map { name => + var idx = -1 + var k = 0 + while (k < fields.size && idx < 0) { if (fields.get(k).getName == name) idx = k; k += 1 } + if (idx < 0) throw new IllegalArgumentException(s"column '$name' not found in Arrow batch") + idx + }.toArray + } + + // Decode one row into a Tuple containing only the projected columns (cheaper + // than getTexeraTuple for wide inputs). `projectedSchema` and `srcIndices` + // (from projectionIndices) align by position. + def getProjectedTuple( + rowIndex: Int, + root: VectorSchemaRoot, + projectedSchema: Schema, + srcIndices: Array[Int] + ): Tuple = { + val vectors = root.getFieldVectors + val values = new Array[Any](srcIndices.length) + var i = 0 + while (i < srcIndices.length) { + val raw = vectors.get(srcIndices(i)).getObject(rowIndex) + values(i) = + try AttributeTypeUtils.parseField(raw, projectedSchema.getAttributes(i).getType) + catch { case _: Exception => null } + i += 1 + } + Tuple.builder(projectedSchema).addSequentially(values).build() + } + /** The wall clock a timestamp column holds, read the way its own field states. * * A Texera TIMESTAMP carries no zone, so the wall clock is the whole of what @@ -232,13 +368,17 @@ object ArrowUtils extends LazyLogging { * @param vectorSchemaRoot The root of the Vectors that stores the Arrow Fields. It contains * multiple Vectors. */ - def setTexeraTuple(tuple: Tuple, index: Int, vectorSchemaRoot: VectorSchemaRoot): Unit = { + def setTexeraTuple(tuple: Tuple, index: Int, vectorSchemaRoot: VectorSchemaRoot): Unit = + setRow(tuple.getFields, index, vectorSchemaRoot) + + // Fill row `index` of the Arrow vectors from a parsed field array (schema order). + def setRow(fields: Array[Any], index: Int, vectorSchemaRoot: VectorSchemaRoot): Unit = { val arrowSchema = vectorSchemaRoot.getSchema val arrowFields = arrowSchema.getFields.asScala.toList for (i <- arrowFields.indices) { val vector: FieldVector = vectorSchemaRoot.getVector(i) - val value = tuple.getField[AnyRef](i) + val value = fields(i) val isNull = value == null arrowFields.apply(i).getFieldType.getType match { case _: ArrowType.Int => diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowIpcRoundTripSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowIpcRoundTripSpec.scala new file mode 100644 index 00000000000..c6b932fb009 --- /dev/null +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/util/ArrowIpcRoundTripSpec.scala @@ -0,0 +1,69 @@ +/* + * 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.texera.amber.util + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.scalatest.flatspec.AnyFlatSpec + +import java.sql.Timestamp + +/** + * Foundation for the Arrow columnar wire format: a batch of tuples must survive + * serializeTuples -> Arrow IPC bytes -> deserializeTuples unchanged. + */ +class ArrowIpcRoundTripSpec extends AnyFlatSpec { + + private val schema: Schema = Schema() + .add(new Attribute("i", AttributeType.INTEGER)) + .add(new Attribute("l", AttributeType.LONG)) + .add(new Attribute("d", AttributeType.DOUBLE)) + .add(new Attribute("s", AttributeType.STRING)) + .add(new Attribute("b", AttributeType.BOOLEAN)) + .add(new Attribute("t", AttributeType.TIMESTAMP)) + + private def tuple(i: Int): Tuple = + Tuple + .builder(schema) + .addSequentially( + Array( + if (i % 10 == 0) null else Int.box(i), + Long.box(i.toLong * 1000L), + Double.box(i * 1.5), + if (i % 7 == 0) null else s"row-$i", + Boolean.box(i % 2 == 0), + new Timestamp(1_600_000_000_000L + i.toLong * 86_400_000L) + ) + ) + .build() + + "Arrow IPC" should "round-trip a batch of tuples unchanged across all types" in { + val n = 5000 + val rows = (0 until n).map(tuple).toArray + val bytes = ArrowUtils.serializeTuples(schema, rows) + val back = ArrowUtils.deserializeTuples(bytes) + assert(back.length == n) + assert(back.sameElements(rows)) + } + + "Arrow IPC" should "round-trip an empty batch" in { + val bytes = ArrowUtils.serializeTuples(schema, Array.empty[Tuple]) + assert(ArrowUtils.deserializeTuples(bytes).isEmpty) + } +} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpExec.scala index 3c26709116f..dd015c4fbc0 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/aggregate/AggregateOpExec.scala @@ -19,8 +19,14 @@ package org.apache.texera.amber.operator.aggregate -import org.apache.texera.amber.core.executor.OperatorExecutor -import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} +import org.apache.arrow.memory.RootAllocator +import org.apache.texera.amber.core.executor.{ + ColumnarOperatorExecutor, + ColumnarResult, + OperatorExecutor +} +import org.apache.texera.amber.core.tuple.{Schema, Tuple, TupleLike} +import org.apache.texera.amber.util.ArrowUtils import org.apache.texera.amber.util.JSONUtils.objectMapper import scala.collection.mutable @@ -28,7 +34,7 @@ import scala.collection.mutable /** * AggregateOpExec performs aggregation operations on input tuples, optionally grouping them by specified keys. */ -class AggregateOpExec(descString: String) extends OperatorExecutor { +class AggregateOpExec(descString: String) extends OperatorExecutor with ColumnarOperatorExecutor { private val desc: AggregateOpDesc = objectMapper.readValue(descString, classOf[AggregateOpDesc]) private var keyedPartialAggregates: mutable.HashMap[List[Object], List[Object]] = _ private var distributedAggregations: List[DistributedAggregation[Object]] = _ @@ -41,6 +47,7 @@ class AggregateOpExec(descString: String) extends OperatorExecutor { override def close(): Unit = { keyedPartialAggregates.clear() distributedAggregations = null + if (columnarAllocator != null) { columnarAllocator.close(); columnarAllocator = null } } override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = { @@ -78,6 +85,38 @@ class AggregateOpExec(descString: String) extends OperatorExecutor { } + // ---- Native-Arrow path: consume the Arrow batch by decoding only the group + // keys and aggregated columns per row, then feeding the same processTuple + // accumulation. Blocking operator, so it emits nothing per batch (Consumed); + // results are produced at onFinish via the row path. + @transient private var columnarAllocator: RootAllocator = _ + @transient private var neededNames: Seq[String] = _ + @transient private var projSchema: Schema = _ + @transient private var projIndices: Array[Int] = _ + + override def processColumnarBatch(arrowIpcBytes: Array[Byte], port: Int): ColumnarResult = { + if (neededNames == null) { + neededNames = (desc.groupByKeys ++ desc.aggregations.flatMap(a => + Option(a.attribute).map(_.trim).filter(_.nonEmpty) + )).distinct + } + if (columnarAllocator == null) columnarAllocator = new RootAllocator() + ArrowUtils.deserializeRootFold(arrowIpcBytes, columnarAllocator) { root => + if (projSchema == null) { + val full = ArrowUtils.toTexeraSchema(root.getSchema) + projSchema = Schema(neededNames.map(full.getAttribute).toList) + projIndices = ArrowUtils.projectionIndices(root, neededNames) + } + val n = root.getRowCount + var i = 0 + while (i < n) { + processTuple(ArrowUtils.getProjectedTuple(i, root, projSchema, projIndices), 0) + i += 1 + } + ColumnarResult.Consumed + } + } + override def onFinish(port: Int): Iterator[TupleLike] = { // Finalize aggregation for all keys and produce the result keyedPartialAggregates.iterator.map { diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpExec.scala index 30f1d955865..6906882e725 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpExec.scala @@ -19,11 +19,225 @@ package org.apache.texera.amber.operator.filter -import org.apache.texera.amber.core.tuple.Tuple +import org.apache.arrow.memory.RootAllocator +import org.apache.arrow.vector.{Float8Vector, IntVector} +import org.apache.texera.amber.core.executor.{ColumnarOperatorExecutor, ColumnarResult} +import org.apache.texera.common.config.ApplicationConfig +import org.apache.texera.amber.core.tuple.{ + AttributeType, + AttributeTypeUtils, + ColumnarBatch, + Tuple, + TupleLike +} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.util.ArrowUtils import org.apache.texera.amber.util.JSONUtils.objectMapper -class SpecializedFilterOpExec(descString: String) extends FilterOpExec { +import java.sql.Timestamp + +class SpecializedFilterOpExec(descString: String) + extends FilterOpExec + with ColumnarOperatorExecutor { private val desc: SpecializedFilterOpDesc = objectMapper.readValue(descString, classOf[SpecializedFilterOpDesc]) + // Row path: OR over predicates, per tuple. setFilterFunc((tuple: Tuple) => desc.predicates.exists(_.evaluate(tuple))) + + private val vectorizedEnabled: Boolean = ApplicationConfig.enableVectorizedOperators + + // Per-predicate row test with per-batch invariants hoisted, resolved from the + // batch schema on first use. null = a predicate shape is not vectorizable, so + // the whole batch falls back to the row path. + @transient private var compiled: Array[Tuple => Boolean] = _ + @transient private var compileAttempted: Boolean = false + + private def cmpOf(c: ComparisonType): Int => Boolean = + c match { + case ComparisonType.GREATER_THAN => (r: Int) => r > 0 + case ComparisonType.GREATER_THAN_OR_EQUAL_TO => (r: Int) => r >= 0 + case ComparisonType.LESS_THAN => (r: Int) => r < 0 + case ComparisonType.LESS_THAN_OR_EQUAL_TO => (r: Int) => r <= 0 + case ComparisonType.EQUAL_TO => (r: Int) => r == 0 + case ComparisonType.NOT_EQUAL_TO => (r: Int) => r != 0 + case _ => null + } + + // Compile one predicate to a Tuple => Boolean matching FilterPredicate.evaluate, + // hoisting the schema lookup and constant parse out of the per-row work. + // Returns null if the shape isn't supported. + private def compileOne(p: FilterPredicate, tpe: AttributeType): Tuple => Boolean = { + val attr = p.attribute + p.condition match { + case ComparisonType.IS_NULL => (t: Tuple) => t.getField[Any](attr) == null + case ComparisonType.IS_NOT_NULL => (t: Tuple) => t.getField[Any](attr) != null + case cond => + val cmp = cmpOf(cond) + if (cmp == null) return null + tpe match { + case AttributeType.INTEGER | AttributeType.DOUBLE => + val c = + try p.value.toDouble + catch { case _: NumberFormatException => return null } + (t: Tuple) => { + val f = t.getField[Any](attr) + if (f == null) false + else cmp(java.lang.Double.compare(f.asInstanceOf[Number].doubleValue(), c)) + } + case AttributeType.LONG => + val c = + try java.lang.Long.valueOf(p.value.trim) + catch { case _: NumberFormatException => return null } + (t: Tuple) => { + val f = t.getField[Any](attr) + if (f == null) false + else cmp(java.lang.Long.compare(f.asInstanceOf[Number].longValue(), c)) + } + case AttributeType.TIMESTAMP => + val c = AttributeTypeUtils.parseTimestamp(p.value.trim).getTime + (t: Tuple) => { + val f = t.getField[Any](attr) + if (f == null) false + else cmp(java.lang.Long.compare(f.asInstanceOf[Timestamp].getTime, c)) + } + case AttributeType.BOOLEAN => + val c = p.value.trim.toLowerCase + (t: Tuple) => { + val f = t.getField[Any](attr) + if (f == null) false else cmp(f.toString.toLowerCase.compareTo(c)) + } + case AttributeType.STRING | AttributeType.ANY => + // Mirror FilterPredicate.evaluateFilterString: numeric compare when + // both field and value parse as double, else lexicographic. + val valNum: java.lang.Double = + try java.lang.Double.valueOf(p.value) + catch { case _: NumberFormatException => null } + (t: Tuple) => { + val f = t.getField[Any](attr) + if (f == null) false + else { + val s = f.toString + if (valNum == null) cmp(s.compareTo(p.value)) + else { + val fd = + try java.lang.Double.valueOf(s.trim) + catch { case _: NumberFormatException => null } + if (fd == null) cmp(s.compareTo(p.value)) + else cmp(java.lang.Double.compare(fd, valNum)) + } + } + } + case _ => null + } + } + } + + private def compileAll(sample: Tuple): Unit = { + compileAttempted = true + val schema = sample.getSchema + val fns = desc.predicates.map { p => + val tpe = + try schema.getAttribute(p.attribute).getType + catch { case _: Throwable => null } + if (tpe == null) null else compileOne(p, tpe) + } + if (fns.nonEmpty && fns.forall(_ != null)) compiled = fns.toArray + } + + override def processBatchMultiPort( + batch: Array[Tuple], + port: Int + ): Option[Iterator[(TupleLike, Option[PortIdentity])]] = { + if (!vectorizedEnabled) return None + if (batch.isEmpty) return Some(Iterator.empty) + if (!compileAttempted) compileAll(batch(0)) + if (compiled == null) return None + + val cb = new ColumnarBatch(batch) + val n = batch.length + val mask = new Array[Boolean](n) + + if (compiled.length == 1 && desc.predicates.size == 1 && isNumericSimd(batch(0))) { + // SIMD fast path: single numeric ordered/eq predicate over a primitive column. + val p = desc.predicates.head + val cmp = cmpOf(p.condition) + val c = p.value.toDouble + val (vals, isNull) = cb.doubleColumn(p.attribute) + var i = 0 + while (i < n) { + mask(i) = !isNull(i) && cmp(java.lang.Double.compare(vals(i), c)) + i += 1 + } + } else { + // General path: OR the compiled per-predicate tests over the rows. + var i = 0 + while (i < n) { + val t = batch(i) + var keep = false + var k = 0 + while (!keep && k < compiled.length) { keep = compiled(k)(t); k += 1 } + mask(i) = keep + i += 1 + } + } + Some(cb.select(mask).map(t => (t, None))) + } + + private def isNumericSimd(sample: Tuple): Boolean = { + val p = desc.predicates.head + if (cmpOf(p.condition) == null) return false + val tpe = + try sample.getSchema.getAttribute(p.attribute).getType + catch { case _: Throwable => return false } + (tpe == AttributeType.INTEGER || tpe == AttributeType.DOUBLE) && + (try { p.value.toDouble; true } + catch { case _: NumberFormatException => false }) + } + + // ---- Native-Arrow path (M2): consume the Arrow batch directly, no tuple decode. + @transient private var columnarAllocator: RootAllocator = _ + + override def processColumnarBatch(arrowIpcBytes: Array[Byte], port: Int): ColumnarResult = { + if (!vectorizedEnabled || desc.predicates.size != 1) return ColumnarResult.Unsupported + val p = desc.predicates.head + val cmp = cmpOf(p.condition) + if (cmp == null) return ColumnarResult.Unsupported + val c = + try p.value.toDouble + catch { case _: NumberFormatException => return ColumnarResult.Unsupported } + if (columnarAllocator == null) columnarAllocator = new RootAllocator() + ArrowUtils.deserializeRootFold(arrowIpcBytes, columnarAllocator) { root => + val fields = root.getSchema.getFields + var fieldIdx = -1 + var k = 0 + while (k < fields.size && fieldIdx < 0) { + if (fields.get(k).getName == p.attribute) fieldIdx = k + k += 1 + } + if (fieldIdx < 0) ColumnarResult.Unsupported + else { + val n = root.getRowCount + val mask = new Array[Boolean](n) + root.getVector(fieldIdx) match { + case v: Float8Vector => + var i = 0 + while (i < n) { + mask(i) = !v.isNull(i) && cmp(java.lang.Double.compare(v.get(i), c)); i += 1 + } + ColumnarResult.Emit(ArrowUtils.selectRows(root, mask, columnarAllocator)) + case v: IntVector => + var i = 0 + while (i < n) { + mask(i) = !v.isNull(i) && cmp(java.lang.Double.compare(v.get(i).toDouble, c)); i += 1 + } + ColumnarResult.Emit(ArrowUtils.selectRows(root, mask, columnarAllocator)) + case _ => ColumnarResult.Unsupported // non-numeric column: fall back to the row path + } + } + } + } + + override def close(): Unit = { + if (columnarAllocator != null) columnarAllocator.close() + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala index a27ae1d5e24..000b1e44127 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/hashJoin/HashJoinProbeOpExec.scala @@ -19,9 +19,16 @@ package org.apache.texera.amber.operator.hashJoin -import org.apache.texera.amber.core.executor.OperatorExecutor -import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} +import org.apache.arrow.memory.RootAllocator +import org.apache.texera.amber.core.executor.{ + ColumnarOperatorExecutor, + ColumnarResult, + OperatorExecutor +} +import org.apache.texera.amber.core.tuple.{Schema, Tuple, TupleLike} +import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc.HASH_JOIN_INTERNAL_KEY_NAME +import org.apache.texera.amber.util.ArrowUtils import org.apache.texera.amber.util.JSONUtils.objectMapper import scala.collection.mutable @@ -62,7 +69,8 @@ object JoinUtils { class HashJoinProbeOpExec[K]( descString: String -) extends OperatorExecutor { +) extends OperatorExecutor + with ColumnarOperatorExecutor { private val desc: HashJoinOpDesc[K] = objectMapper.readValue(descString, classOf[HashJoinOpDesc[K]]) @@ -74,6 +82,44 @@ class HashJoinProbeOpExec[K]( override def close(): Unit = { buildTableHashMap.clear() + if (columnarAllocator != null) { columnarAllocator.close(); columnarAllocator = null } + } + + // ---- Native-Arrow probe: on the probe port, decode only the probe key column + // to test each row against the build map, and fully decode a row only when it + // matches (or on an outer join). For a selective inner join, most probe rows + // are never fully decoded. The build-load port (0) needs all columns, so it + // falls back to the row path. + @transient private var columnarAllocator: RootAllocator = _ + @transient private var keySchema: Schema = _ + + override def processColumnarBatch(arrowIpcBytes: Array[Byte], port: Int): ColumnarResult = { + if (port == 0) return ColumnarResult.Unsupported + val isOuter = desc.joinType == JoinType.RIGHT_OUTER || desc.joinType == JoinType.FULL_OUTER + if (columnarAllocator == null) columnarAllocator = new RootAllocator() + ArrowUtils.deserializeRootFold(arrowIpcBytes, columnarAllocator) { root => + if (keySchema == null) { + val full = ArrowUtils.toTexeraSchema(root.getSchema) + keySchema = Schema(List(full.getAttribute(desc.probeAttributeName))) + } + val keyIdx = ArrowUtils.projectionIndices(root, Seq(desc.probeAttributeName)) + val out = new ListBuffer[(TupleLike, Option[PortIdentity])] + val n = root.getRowCount + var i = 0 + while (i < n) { + val key = ArrowUtils + .getProjectedTuple(i, root, keySchema, keyIdx) + .getField(desc.probeAttributeName) + .asInstanceOf[K] + val matched = buildTableHashMap.get(key).exists(_._1.nonEmpty) + if (matched || isOuter) { + // Reuse the exact row-path join logic (also marks the build side joined). + processTuple(ArrowUtils.getTexeraTuple(i, root), 1).foreach(t => out += ((t, None))) + } + i += 1 + } + ColumnarResult.EmitRows(out.iterator) + } } override def processTuple(tuple: Tuple, port: Int): Iterator[TupleLike] = diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala index b0512cc826a..8d0760671d6 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpExec.scala @@ -20,15 +20,19 @@ package org.apache.texera.amber.operator.projection import com.google.common.base.Preconditions -import org.apache.texera.amber.core.tuple.{Tuple, TupleLike} +import org.apache.arrow.memory.RootAllocator +import org.apache.texera.amber.core.executor.{ColumnarOperatorExecutor, ColumnarResult} +import org.apache.texera.amber.core.tuple.{Attribute, Schema, Tuple, TupleLike} import org.apache.texera.amber.operator.map.MapOpExec +import org.apache.texera.amber.util.ArrowUtils import org.apache.texera.amber.util.JSONUtils.objectMapper import scala.collection.mutable class ProjectionOpExec( descString: String -) extends MapOpExec { +) extends MapOpExec + with ColumnarOperatorExecutor { val desc: ProjectionOpDesc = objectMapper.readValue(descString, classOf[ProjectionOpDesc]) setMapFunc(project) @@ -66,4 +70,35 @@ class ProjectionOpExec( TupleLike(fields.toSeq: _*) } + // ---- Native-Arrow path: select (and rename) columns directly on the Arrow + // batch, no per-row Tuple decode. Streaming 1:1, so it emits the projected batch. + @transient private var columnarAllocator: RootAllocator = _ + + override def processColumnarBatch(arrowIpcBytes: Array[Byte], port: Int): ColumnarResult = { + Preconditions.checkArgument(desc.attributes.nonEmpty) + if (columnarAllocator == null) columnarAllocator = new RootAllocator() + ArrowUtils.deserializeRootFold(arrowIpcBytes, columnarAllocator) { root => + val full = ArrowUtils.toTexeraSchema(root.getSchema) + // (originalName, alias) in output order, matching the row `project` path. + val selected: List[(String, String)] = + if (desc.isDrop) { + val drop = desc.attributes.map(_.getOriginalAttribute).toSet + full.getAttributeNames.filterNot(drop.contains).map(a => (a, a)) + } else desc.attributes.map(u => (u.getOriginalAttribute, u.getAlias)) + val aliases = selected.map(_._2) + if (aliases.distinct.size != aliases.size) { + throw new RuntimeException("have duplicated attribute name/alias") + } + val outSchema = Schema(selected.map { + case (orig, alias) => new Attribute(alias, full.getAttribute(orig).getType) + }) + val srcIdx = ArrowUtils.projectionIndices(root, selected.map(_._1)) + ColumnarResult.Emit(ArrowUtils.selectColumns(root, srcIdx, outSchema, columnarAllocator)) + } + } + + override def close(): Unit = { + if (columnarAllocator != null) { columnarAllocator.close(); columnarAllocator = null } + } + } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala index 147238536d8..22a18bf7541 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala @@ -21,9 +21,12 @@ package org.apache.texera.amber.operator.source.scan.csv import com.univocity.parsers.common.TextParsingException import com.univocity.parsers.csv.{CsvFormat, CsvParser, CsvParserSettings} +import org.apache.arrow.memory.RootAllocator +import org.apache.arrow.vector.VectorSchemaRoot import org.apache.texera.amber.core.executor.SourceOperatorExecutor import org.apache.texera.amber.core.storage.DocumentFactory import org.apache.texera.amber.core.tuple.{AttributeTypeUtils, Schema, TupleLike} +import org.apache.texera.amber.util.ArrowUtils import org.apache.texera.amber.util.JSONUtils.objectMapper import org.apache.texera.dao.SiteSettings @@ -40,6 +43,8 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat var numRowGenerated = 0 private var maxColumns: Int = CSVScanSourceOpExec.DEFAULT_MAX_COLUMNS private val schema: Schema = desc.sourceSchema() + private var columnarAllocator: RootAllocator = _ + private val columnarBatchRows = 10000 override def produceTuple(): Iterator[TupleLike] = { @@ -80,6 +85,37 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat tupleIterator } + // Columnar produce: parse CSV straight into Arrow batches (no per-row Tuple). + override def produceColumnarBatch(): Option[Iterator[VectorSchemaRoot]] = { + columnarAllocator = new RootAllocator() + var toSkip = desc.offset.getOrElse(0) + while (toSkip > 0) { CSVScanSourceOpExec.parseNextRow(parser, maxColumns); toSkip -= 1 } + val limit = desc.limit.getOrElse(Int.MaxValue) + Some(new Iterator[VectorSchemaRoot] { + private var pending: Array[String] = CSVScanSourceOpExec.parseNextRow(parser, maxColumns) + private var emitted = 0 + + override def hasNext: Boolean = pending != null && emitted < limit + + override def next(): VectorSchemaRoot = { + val root = VectorSchemaRoot.create(ArrowUtils.fromTexeraSchema(schema), columnarAllocator) + var filled = 0 + while (pending != null && filled < columnarBatchRows && emitted < limit) { + try { + val fields = AttributeTypeUtils.parseFields(pending.asInstanceOf[Array[Any]], schema) + ArrowUtils.setRow(fields, filled, root) + filled += 1 + } catch { case _: Throwable => } // skip malformed row (matches row path) + emitted += 1 + numRowGenerated += 1 + pending = CSVScanSourceOpExec.parseNextRow(parser, maxColumns) + } + root.setRowCount(filled) + root + } + }) + } + override def open(): Unit = { inputReader = new InputStreamReader( DocumentFactory.openReadonlyDocument(new URI(desc.fileName.get)).asInputStream(), @@ -110,6 +146,9 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat if (inputReader != null) { inputReader.close() } + if (columnarAllocator != null) { + columnarAllocator.close() + } } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/VectorizedFilterCorrectnessSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/VectorizedFilterCorrectnessSpec.scala new file mode 100644 index 00000000000..37f675ab54f --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/VectorizedFilterCorrectnessSpec.scala @@ -0,0 +1,236 @@ +/* + * 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.texera.amber.operator.filter + +import org.apache.texera.amber.core.executor.ColumnarResult +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.util.ArrowUtils +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.scalatest.flatspec.AnyFlatSpec + +import java.sql.Timestamp +import scala.util.Random + +/** + * Lever 8.2 correctness: the vectorized batch path (processBatchMultiPort) must + * return exactly the same surviving tuples as the row path (the stock + * FilterPredicate), across every attribute type, every comparison operator, + * nulls, IS NULL / IS NOT NULL, and multi-predicate OR. + */ +class VectorizedFilterCorrectnessSpec extends AnyFlatSpec { + + private val orderedOps = Seq( + ComparisonType.GREATER_THAN, + ComparisonType.GREATER_THAN_OR_EQUAL_TO, + ComparisonType.LESS_THAN, + ComparisonType.LESS_THAN_OR_EQUAL_TO, + ComparisonType.EQUAL_TO, + ComparisonType.NOT_EQUAL_TO + ) + + private def execFor(preds: (String, ComparisonType, String)*): SpecializedFilterOpExec = { + val desc = new SpecializedFilterOpDesc() + desc.predicates = preds.map { case (a, c, v) => new FilterPredicate(a, c, v) }.toList + new SpecializedFilterOpExec(objectMapper.writeValueAsString(desc)) + } + + private def build(schema: Schema, attr: String, values: Seq[Any]): Array[Tuple] = + values.map(v => Tuple.builder(schema).add(schema.getAttribute(attr), v).build()).toArray + + // For each operator, assert vectorized survivors == row survivors AND the fast path ran. + private def checkAllOps(schema: Schema, rows: Array[Tuple], attr: String, value: String): Unit = + orderedOps.foreach { op => + val exec = execFor((attr, op, value)) + val row = rows.filter(exec.filterFunc).toList + val vec = exec.processBatchMultiPort(rows, 0) + assert(vec.isDefined, s"expected vectorized path for $op on $attr") + assert(vec.get.map(_._1).toList == row, s"mismatch: op=$op attr=$attr value=$value") + } + + private def col(name: String, t: AttributeType) = Schema().add(new Attribute(name, t)) + + "Vectorized filter" should "match the row path on DOUBLE (with nulls)" in { + val s = col("v", AttributeType.DOUBLE) + val rng = new Random(1) + val rows = build( + s, + "v", + (0 until 4000).map(i => + if (i % 17 == 0) null else Double.box((rng.nextInt(100) - 10).toDouble) + ) + ) + checkAllOps(s, rows, "v", "25") + } + + "Vectorized filter" should "match the row path on INTEGER (with nulls)" in { + val s = col("v", AttributeType.INTEGER) + val rng = new Random(2) + val rows = + build(s, "v", (0 until 4000).map(i => if (i % 13 == 0) null else Int.box(rng.nextInt(60)))) + checkAllOps(s, rows, "v", "30") + } + + "Vectorized filter" should "match the row path on LONG (with nulls)" in { + val s = col("v", AttributeType.LONG) + val rng = new Random(3) + val rows = build( + s, + "v", + (0 until 4000).map(i => if (i % 11 == 0) null else Long.box(rng.nextInt(1000).toLong)) + ) + checkAllOps(s, rows, "v", "500") + } + + "Vectorized filter" should "match the row path on STRING, numeric value (coercion)" in { + val s = col("v", AttributeType.STRING) + val rng = new Random(4) + // mix of numeric-looking and non-numeric strings + val rows = build( + s, + "v", + (0 until 4000).map { i => + if (i % 19 == 0) null + else if (i % 3 == 0) s"item-${rng.nextInt(50)}" + else rng.nextInt(50).toString + } + ) + checkAllOps(s, rows, "v", "25") + } + + "Vectorized filter" should "match the row path on STRING, non-numeric value (lexicographic)" in { + val s = col("v", AttributeType.STRING) + val rng = new Random(5) + val rows = build( + s, + "v", + (0 until 3000).map(i => + if (i % 23 == 0) null else s"${('A' + rng.nextInt(26)).toChar}${rng.nextInt(10)}" + ) + ) + checkAllOps(s, rows, "v", "M5") + } + + "Vectorized filter" should "match the row path on BOOLEAN (with nulls)" in { + val s = col("v", AttributeType.BOOLEAN) + val rng = new Random(6) + val rows = build( + s, + "v", + (0 until 2000).map(i => if (i % 7 == 0) null else Boolean.box(rng.nextBoolean())) + ) + Seq(ComparisonType.EQUAL_TO, ComparisonType.NOT_EQUAL_TO).foreach { op => + val exec = execFor(("v", op, "true")) + assert( + exec.processBatchMultiPort(rows, 0).get.map(_._1).toList == rows + .filter(exec.filterFunc) + .toList + ) + } + } + + "Vectorized filter" should "match the row path on TIMESTAMP (with nulls)" in { + val s = col("v", AttributeType.TIMESTAMP) + val base = 1_600_000_000_000L + val rows = build( + s, + "v", + (0 until 2000).map(i => + if (i % 9 == 0) null else new Timestamp(base + i.toLong * 86_400_000L) + ) + ) + checkAllOps(s, rows, "v", "2020-10-01 00:00:00") + } + + "Vectorized filter" should "match the row path for IS NULL / IS NOT NULL" in { + val s = col("v", AttributeType.DOUBLE) + val rows = + build(s, "v", (0 until 2000).map(i => if (i % 5 == 0) null else Double.box(i.toDouble))) + Seq(ComparisonType.IS_NULL, ComparisonType.IS_NOT_NULL).foreach { op => + val exec = execFor(("v", op, "")) + assert( + exec.processBatchMultiPort(rows, 0).get.map(_._1).toList == rows + .filter(exec.filterFunc) + .toList + ) + } + } + + "Vectorized filter" should "match the row path for multi-predicate OR" in { + val s = Schema() + .add(new Attribute("a", AttributeType.DOUBLE)) + .add(new Attribute("b", AttributeType.INTEGER)) + val rng = new Random(8) + val rows = (0 until 3000).map { i => + val a: Any = if (i % 15 == 0) null else Double.box((rng.nextInt(100)).toDouble) + val b: Any = if (i % 21 == 0) null else Int.box(rng.nextInt(100)) + Tuple.builder(s).add(s.getAttribute("a"), a).add(s.getAttribute("b"), b).build() + }.toArray + val exec = + execFor(("a", ComparisonType.GREATER_THAN, "80"), ("b", ComparisonType.LESS_THAN, "10")) + assert( + exec.processBatchMultiPort(rows, 0).get.map(_._1).toList == rows + .filter(exec.filterFunc) + .toList + ) + } + + "Vectorized filter" should "fall back (None) when a numeric value cannot be parsed" in { + val s = col("v", AttributeType.DOUBLE) + val rows = build(s, "v", Seq(Double.box(1.0), Double.box(2.0))) + val exec = execFor(("v", ComparisonType.GREATER_THAN, "not-a-number")) + assert(exec.processBatchMultiPort(rows, 0).isEmpty) + } + + "Native-Arrow filter" should "match the row path via processColumnarBatch (M2)" in { + Seq(("v", AttributeType.DOUBLE, "50"), ("w", AttributeType.INTEGER, "40")).foreach { + case (name, tpe, value) => + val s = col(name, tpe) + val rng = new Random(99) + val rows = build( + s, + name, + (0 until 4000).map { i => + if (i % 17 == 0) null + else if (tpe == AttributeType.DOUBLE) Double.box((rng.nextInt(100)).toDouble) + else Int.box(rng.nextInt(100)) + } + ) + orderedOps.foreach { op => + val exec = execFor((name, op, value)) + val bytes = ArrowUtils.serializeTuples(s, rows) + val resultRoot = exec.processColumnarBatch(bytes, 0) match { + case ColumnarResult.Emit(r) => r + case other => fail(s"expected Emit, got $other") + } + val survivors = + (0 until resultRoot.getRowCount) + .map(i => ArrowUtils.getTexeraTuple(i, resultRoot)) + .toList + resultRoot.close() + exec.close() + val rowSurvivors = rows.filter(exec.filterFunc).toList + assert( + survivors.map(_.getField[Any](name)) == rowSurvivors.map(_.getField[Any](name)), + s"native-arrow mismatch: $name $op $value" + ) + } + } + } +} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala index 1d8ee9717bb..947197f825d 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpExecSpec.scala @@ -19,8 +19,10 @@ package org.apache.texera.amber.operator.projection +import org.apache.texera.amber.core.executor.ColumnarResult import org.apache.texera.amber.core.tuple._ import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.util.ArrowUtils import org.apache.texera.amber.util.JSONUtils.objectMapper import org.scalatest.BeforeAndAfter import org.scalatest.flatspec.AnyFlatSpec @@ -130,6 +132,53 @@ class ProjectionOpExecSpec extends AnyFlatSpec with BeforeAndAfter { } } + private def batch(n: Int): Array[Tuple] = + (0 until n).map { i => + Tuple + .builder(tupleSchema) + .add(new Attribute("field1", AttributeType.STRING), s"s$i") + .add(new Attribute("field2", AttributeType.INTEGER), Int.box(i)) + .add(new Attribute("field3", AttributeType.BOOLEAN), Boolean.box(i % 2 == 0)) + .build() + }.toArray + + it should "match the row path via processColumnarBatch (rename + reorder)" in { + val d = new ProjectionOpDesc() + d.attributes = List(new AttributeUnit("field2", "f2"), new AttributeUnit("field1", "f1")) + val exec = new ProjectionOpExec(objectMapper.writeValueAsString(d)) + exec.open() + val rows = batch(200) + val out = exec.processColumnarBatch(ArrowUtils.serializeTuples(tupleSchema, rows), 0) match { + case ColumnarResult.Emit(r) => r + case other => fail(s"expected Emit, got $other") + } + val cols = out.getSchema.getFields + assert(cols.get(0).getName == "f2" && cols.get(1).getName == "f1") + val decoded = (0 until out.getRowCount).map(i => ArrowUtils.getTexeraTuple(i, out)).toList + out.close(); exec.close() + assert(decoded.map(_.getField[Any]("f2")) == rows.map(_.getField[Any]("field2")).toList) + assert(decoded.map(_.getField[Any]("f1")) == rows.map(_.getField[Any]("field1")).toList) + } + + it should "match the row path via processColumnarBatch (drop mode)" in { + val d = new ProjectionOpDesc() + d.isDrop = true + d.attributes = List(new AttributeUnit("field2", "field2")) + val exec = new ProjectionOpExec(objectMapper.writeValueAsString(d)) + exec.open() + val rows = batch(150) + val out = exec.processColumnarBatch(ArrowUtils.serializeTuples(tupleSchema, rows), 0) match { + case ColumnarResult.Emit(r) => r + case other => fail(s"expected Emit, got $other") + } + val names = (0 until out.getSchema.getFields.size).map(out.getSchema.getFields.get(_).getName) + assert(names == Seq("field1", "field3")) + val decoded = (0 until out.getRowCount).map(i => ArrowUtils.getTexeraTuple(i, out)).toList + out.close(); exec.close() + assert(decoded.map(_.getField[Any]("field1")) == rows.map(_.getField[Any]("field1")).toList) + assert(decoded.map(_.getField[Any]("field3")) == rows.map(_.getField[Any]("field3")).toList) + } + it should "allow empty alias" in { opDesc.attributes = List( new AttributeUnit("field2", "f2"),