feat: Add ChangelogRowVector for RowKind propagation

velox/experimental/stateful/CMakeLists.txt

@@ -37,7 +37,8 @@ velox_add_library(
   GroupWindowAggregator.cpp
   WatermarkGenerator.cpp
   WatermarkSource.cpp
-  WatermarkAssigner.cpp)
+  WatermarkAssigner.cpp
+  ChangelogRowVector.cpp)
 
 velox_link_libraries(
   velox_stateful_exec

velox/experimental/stateful/ChangelogRowVector.cpp

@@ -0,0 +1,101 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "velox/experimental/stateful/ChangelogRowVector.h"
+
+#include "velox/common/base/Exceptions.h"
+#include "velox/type/Type.h"
+#include "velox/vector/ConstantVector.h"
+
+namespace facebook::velox::stateful {
+
+ChangelogRowVector::ChangelogRowVector(
+    RowVectorPtr value,
+    SimpleVectorPtr<int8_t> rowKind)
+    : value_(std::move(value)), rowKind_(std::move(rowKind)) {
+  VELOX_CHECK_NOT_NULL(value_);
+  if (rowKind_ == nullptr) {
+    appendOnly_ = true;
+    return;
+  }
+  VELOX_USER_CHECK_EQ(
+      rowKind_->size(),
+      value_->size(),
+      "rowKind size {} must match value size {}",
+      rowKind_->size(),
+      value_->size());
+  appendOnly_ = rowKind_->isConstantEncoding() &&
+      rowKind_->as<ConstantVector<int8_t>>()->valueAt(0) ==
+          static_cast<int8_t>(RowKind::INSERT);
+}
+
+ChangelogRowVectorPtr ChangelogRowVector::fromMergedRowVector(
+    RowVectorPtr merged) {
+  VELOX_USER_CHECK_NOT_NULL(merged, "merged RowVector must not be null");
+  auto rowType = asRowType(merged->type());
+  if (rowType->size() == 0 ||
+      rowType->nameOf(rowType->size() - 1) != kRowKindColumnName) {
+    return std::make_shared<ChangelogRowVector>(std::move(merged));
+  }
+  auto lastIdx = rowType->size() - 1;
+  auto kindVector = merged->childAt(lastIdx);
+  auto simpleKind = std::dynamic_pointer_cast<SimpleVector<int8_t>>(kindVector);
+  VELOX_USER_CHECK_NOT_NULL(
+      simpleKind,
+      "$row_kind column must be a SimpleVector<int8_t>, got encoding: {}",
+      kindVector->encoding());
+  std::vector<std::string> names;
+  std::vector<TypePtr> types;
+  std::vector<VectorPtr> children;
+  names.reserve(lastIdx);
+  types.reserve(lastIdx);
+  children.reserve(lastIdx);
+  for (size_t i = 0; i < lastIdx; ++i) {
+    names.emplace_back(rowType->nameOf(i));
+    types.emplace_back(rowType->childAt(i));
+    children.emplace_back(merged->childAt(i));
+  }
+  auto valueType = ROW(std::move(names), std::move(types));
+  auto value = std::make_shared<RowVector>(
+      merged->pool(), valueType, nullptr, merged->size(), std::move(children));
+  return std::make_shared<ChangelogRowVector>(std::move(value), simpleKind);
+}
+
+RowVectorPtr ChangelogRowVector::toMergedRowVector() const {
+  if (appendOnly_) {
+    return value_;
+  }
+  auto rowType = asRowType(value_->type());
+  std::vector<std::string> names;
+  std::vector<TypePtr> types;
+  std::vector<VectorPtr> children;
+  auto n = rowType->size();
+  names.reserve(n + 1);
+  types.reserve(n + 1);
+  children.reserve(n + 1);
+  for (size_t i = 0; i < n; ++i) {
+    names.emplace_back(rowType->nameOf(i));
+    types.emplace_back(rowType->childAt(i));
+    children.emplace_back(value_->childAt(i));
+  }
+  names.emplace_back(kRowKindColumnName);
+  types.emplace_back(TINYINT());
+  children.emplace_back(rowKind_);
+  auto mergedType = ROW(std::move(names), std::move(types));
+  return std::make_shared<RowVector>(
+      value_->pool(), mergedType, nullptr, size(), std::move(children));
+}
+
+} // namespace facebook::velox::stateful

velox/experimental/stateful/ChangelogRowVector.h

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include "velox/experimental/stateful/RowKind.h"
+#include "velox/vector/ComplexVector.h"
+#include "velox/vector/SimpleVector.h"
+
+namespace facebook::velox::stateful {
+
+// Forward declaration so the class body can reference the shared_ptr alias.
+class ChangelogRowVector;
+using ChangelogRowVectorPtr = std::shared_ptr<ChangelogRowVector>;
+
+// Combines a RowVector (schema columns) with a per-row RowKind vector for
+// carrying changelog semantics.
+//
+// RowKind ordinals align with Flink org.apache.flink.types.RowKind:
+//   0 = INSERT (+I), 1 = UPDATE_BEFORE (-U),
+//   2 = UPDATE_AFTER (+U), 3 = DELETE (-D).
+//
+// appendOnly is inferred from rowKind: true when rowKind is null or is a
+// Constant(INSERT) vector.
+class ChangelogRowVector {
+ public:
+  // rowKind may be null (appendOnly, all INSERT) or non-null with size ==
+  // value->size().
+  explicit ChangelogRowVector(
+      RowVectorPtr value,
+      SimpleVectorPtr<int8_t> rowKind = nullptr);
+
+  const RowVectorPtr& value() const {
+    return value_;
+  }
+
+  // May be null when appendOnly is true.
+  const SimpleVectorPtr<int8_t>& rowKind() const {
+    return rowKind_;
+  }
+
+  vector_size_t size() const {
+    return value_->size();
+  }
+
+  bool appendOnly() const {
+    return appendOnly_;
+  }
+
+  // Splits a merged RowVector (user columns + optional trailing $row_kind) into
+  // a ChangelogRowVector. If $row_kind is absent, the result is appendOnly.
+  static ChangelogRowVectorPtr fromMergedRowVector(RowVectorPtr merged);
+
+  // Merges back into a RowVector. If appendOnly, returns value_ as-is.
+  // Otherwise appends rowKind_ as the trailing $row_kind column.
+  RowVectorPtr toMergedRowVector() const;
+
+ private:
+  RowVectorPtr value_;
+  SimpleVectorPtr<int8_t> rowKind_;
+  bool appendOnly_;
+};
+
+} // namespace facebook::velox::stateful

velox/experimental/stateful/GroupWindowAggregator.cpp

@@ -75,7 +75,8 @@ void GroupWindowAggregator::initializeState() {
 void GroupWindowAggregator::addInput(StreamElementPtr input) {
   VELOX_CHECK(!input_, "Last input has not been processed");
   auto record = std::static_pointer_cast<StreamRecord>(input);
-  input_ = record->record();
+  // TODO: upgrade input_ to ChangelogRowVector to preserve rowKind.
+  input_ = record->record()->value();
 }
 
 void GroupWindowAggregator::advance() {

velox/experimental/stateful/LocalWindowAggregator.cpp

@@ -41,7 +41,8 @@ LocalWindowAggregator::LocalWindowAggregator(
 void LocalWindowAggregator::addInput(StreamElementPtr input) {
   VELOX_CHECK(!input_, "Last input has not been processed");
   auto record = std::static_pointer_cast<StreamRecord>(input);
-  input_ = record->record();
+  // TODO: upgrade input_ to ChangelogRowVector to preserve rowKind.
+  input_ = record->record()->value();
 }
 
 void LocalWindowAggregator::advance() {

velox/experimental/stateful/RowKind.h

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include <cstdint>
+#include <string_view>
+
+namespace facebook::velox::stateful {
+
+enum class RowKind : int8_t {
+  INSERT = 0,
+  UPDATE_BEFORE = 1,
+  UPDATE_AFTER = 2,
+  DELETE = 3,
+};
+
+// Name of the synthetic trailing column that carries RowKind across the JNI
+// boundary inside a merged RowVector.
+constexpr std::string_view kRowKindColumnName = "$row_kind";
+
+} // namespace facebook::velox::stateful

velox/experimental/stateful/StatefulOperator.cpp

@@ -52,8 +52,9 @@ bool StatefulOperator::isFinished() {
 
 void StatefulOperator::addInput(StreamElementPtr input) {
   auto record = std::static_pointer_cast<StreamRecord>(input);
-  operator_->traceInput(record->record());
-  operator_->addInput(record->record());
+  const auto& chrv = record->record();
+  operator_->traceInput(chrv->value());
+  operator_->addInput(chrv->value());
 }
 
 bool StatefulOperator::sourceEmpty() {

velox/experimental/stateful/StatefulOperator.h

@@ -47,15 +47,26 @@ class StatefulOperator {
       : keyedStateBackendParameters_(keyedStateBackendParameters),
         operator_(std::move(op)),
         targets_(std::move(targets)) {
-    sink = operator_->operatorType() == "TableWrite";
+    sink_ = operator_->operatorType() == "TableWrite";
   }
 
+  virtual ~StatefulOperator() = default;
+
   virtual void initialize();
 
   virtual bool isFinished();
 
+  // Receives a StreamRecord and feeds value() to the underlying velox operator.
+  // The base implementation drops rowKind, which is correct for stateless
+  // passthrough. Operators with retract semantics (group aggregation, stream
+  // join, etc.) must override this and advance() to act on rowKind.
   virtual void addInput(StreamElementPtr input);
 
+  // Pulls the next output batch from the underlying velox operator and wraps it
+  // as a StreamRecord. The base implementation assumes appendOnly output (no
+  // $row_kind column), which matches stateless passthrough. Operators with
+  // retract semantics must override this together with addInput() to emit
+  // rowKind according to their changelog semantics.
   virtual void advance();
 
   void advanceWithFuture(ContinueFuture* future);
@@ -152,7 +163,7 @@ class StatefulOperator {
 
  private:
   bool isSink() {
-    return sink;
+    return sink_;
   }
 
   bool isSource() const {
@@ -161,7 +172,7 @@ class StatefulOperator {
 
   std::unique_ptr<exec::Operator> operator_;
   std::vector<std::unique_ptr<StatefulOperator>> targets_;
-  bool sink;
+  bool sink_;
   bool sourceEmpty_ = true;
   std::unique_ptr<CombinedWatermarkStatus> combinedWatermarkStatus_;
   StreamOperatorStateHandlerPtr stateHandler_;

velox/experimental/stateful/StatefulPlanner.cpp

@@ -45,6 +45,7 @@
 #include "velox/experimental/stateful/KeySelector.h"
 #include "velox/experimental/stateful/LocalWindowAggregator.h"
 #include "velox/experimental/stateful/StatefulPlanNode.h"
+#include "velox/experimental/stateful/StatefulSourceOperator.h"
 #include "velox/experimental/stateful/StreamJoin.h"
 #include "velox/experimental/stateful/StreamKeyedOperator.h"
 #include "velox/experimental/stateful/StreamPartition.h"
@@ -398,6 +399,10 @@ StatefulOperatorPtr StatefulPlanner::transformGenericOperator(
         std::move(op), std::move(targets), std::move(watermarkGenerator));
   }
   std::unique_ptr<exec::Operator> op = transformOperator(planNode.node());
+  if (std::dynamic_pointer_cast<const core::TableScanNode>(planNode.node())) {
+    return std::make_unique<StatefulSourceOperator>(
+        std::move(op), std::move(targets));
+  }
   return std::make_unique<StatefulOperator>(std::move(op), std::move(targets));
 }
 

velox/experimental/stateful/StatefulSourceOperator.h

@@ -0,0 +1,52 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include "velox/experimental/stateful/ChangelogRowVector.h"
+#include "velox/experimental/stateful/StatefulOperator.h"
+#include "velox/experimental/stateful/StreamElement.h"
+
+namespace facebook::velox::stateful {
+
+// Wraps a source operator (TableScan, etc.) that has no upstream input. Each
+// RowVector from the underlying source is wrapped as an appendOnly
+// ChangelogRowVector (rowKind = null), since connector sources emit only
+// INSERT rows. Future CDC-style sources that emit native RowKind should
+// introduce a separate subclass.
+class StatefulSourceOperator : public StatefulOperator {
+ public:
+  using StatefulOperator::StatefulOperator;
+
+  // Sources never receive input from upstream. Reject loudly to catch wiring
+  // bugs in the operator chain.
+  void addInput(StreamElementPtr /*input*/) final {
+    VELOX_FAIL("StatefulSourceOperator does not support addInput");
+  }
+
+  void advance() override {
+    setSourceEmpty(true);
+    auto intermediateResult = op()->getOutput();
+    if (!intermediateResult) {
+      return;
+    }
+    setSourceEmpty(false);
+    auto crv =
+        ChangelogRowVector::fromMergedRowVector(std::move(intermediateResult));
+    pushOutput(std::make_shared<StreamRecord>(getPlanNodeId(), std::move(crv)));
+  }
+};
+
+} // namespace facebook::velox::stateful