[native] Operators for unsaferow shuffle #18327

beroyfb · 2022-09-12T05:21:39Z

The new operators include:

PartitionAndSerialize which partitions
and also serializes rows in the row vector as a vector of string (each string holding a serialized row). It uses hive partition function. Later we may add murmur3 hash as well.
ShuffleWrite which writes the result of serialized vectors into an external shuffle system
Extension of ExchangeSource that receives serialized unsaferow blocks and outputs row vectors after deserialization
An interface for an external shuffle

Test plan - presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp is an end-2-end test including a basic shuffle and all the new operators

== NO RELEASE NOTE ==

linux-foundation-easycla · 2022-09-12T05:21:41Z

The committers listed above are authorized under a signed CLA.

✅ login: beroyfb (4d81db3a5c57397247069fce2b452bf0036b4270)

mbasmanova

@beroyfb Looks good overall. This is a large change with many separate pieces. It would be nice to break it up into multiple commits. Perhaps,

Add PartitionAndSerialize operator.
Add UnsafeRowExchangeSource.
Add ShuffleInterface.
Add ShuffleWrite operator.

mbasmanova · 2022-09-12T20:02:54Z

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.cpp

+
+    // Initializing hive partition function
+    auto numPartitions = planNode->numPartitions();
+    VELOX_CHECK(!planNode->keys().empty(), "Empty keys for hive hash!");


Perhaps, move this check to plan node constructor to fail earlier.

mbasmanova · 2022-09-12T20:03:13Z

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.cpp

+    auto inputType = planNode->sources()[0]->outputType();
+    auto keyChannels = toChannels(inputType, planNode->keys());
+
+    // Initializing hive partition function


Comments should be full sentences, e.g. start with a capital letter and end with a period. Please, fix throughout the PR.

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.cpp

mbasmanova · 2022-09-12T20:05:07Z

presto-native-execution/presto_cpp/main/operators/ShuffleInterface.h

+  // Write to the shuffle one row at a time
+  virtual void collect(int32_t partition, std::string_view data) = 0;
+  // Tell the shuffle system the writer is done
+  virtual void noMoreData() = 0;


Don't we need to pass a boolean flag indicating whether we are closing normally or aborting on error?

Good point. You are right.

mbasmanova · 2022-09-12T20:05:24Z

presto-native-execution/presto_cpp/main/operators/ShuffleInterface.h

+ public:
+  // Write to the shuffle one row at a time
+  virtual void collect(int32_t partition, std::string_view data) = 0;
+  // Tell the shuffle system the writer is done


Add empty lines between methods.

mbasmanova · 2022-09-12T20:06:38Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+}
+} // namespace
+
+class UnsaferowShuffleTest : public exec::test::OperatorTestBase {


typo: Unsaferow -> UnsafeRow

Still needs to be fixed.

mbasmanova · 2022-09-12T20:07:46Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+    }
+
+    for (auto& result : serializedResults) {
+      std::cout << "Partition: " << i << ": " << result->toString()


Remove print statements from the tests.

mbasmanova · 2022-09-12T20:09:20Z

presto-native-execution/presto_cpp/main/operators/UnsaferowExchangeSource.h

+
+namespace facebook::presto::operators {
+
+class UnsafeRowExchangeSource : public velox::exec::ExchangeSource {


Typo in the file name: Unsaferow -> UnsafeRow.

mbasmanova · 2022-09-12T20:32:36Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+  auto plan =
+      exec::test::PlanBuilder()
+          .values({data}, true)
+          .addNode(


use addPartitionAndSerializeNode

mbasmanova · 2022-09-12T20:34:09Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+    return rowVec;
+  }
+
+  void assertEqualVectors(const VectorPtr& expected, const VectorPtr& actual) {


Can we use the existing function from velox/vector/tests/VectorTestBase.h ?

I tried but given that it doesn't seem to be reference-able in presto_cpp

doesn't seem to be reference-able in presto_cpp

Hmm... what does that mean? What happens if you try using this function?

The body of assertEqualVectors is located under VeloxVectorTest.cpp which is under velox_vector_test_lib. When I add that lib to CMake file I get a link error:

ld: library not found for -lvelox_vector_test_lib clang: error: linker command failed with exit code 1 (use -v to see invocation)

Aha... This needs to be fixed.

In Velox, there are 2 build variables that control what test code is built: VELOX_BUILD_TESTING and VELOX_BUILD_TEST_UTILS. VELOX_BUILD_TESTING is intended to allow for disabling building of tests (to speed up builds). VELOX_BUILD_TEST_UTILS is intended to allow for disabling building test utilities.

VectorTestBase.* files are test utilities, however, their build is guarded by VELOX_BUILD_TESTING, not VELOX_BUILD_TEST_UTILS.

presto-native-execution/velox/velox/vector/CMakeLists.txt

if(${VELOX_BUILD_TESTING} OR ${VELOX_ENABLE_BENCHMARKS_BASIC}) add_subdirectory(tests) add_subdirectory(benchmarks) endif()

This is incorrect as we should check VELOX_BUILD_TEST_UTILS variable when deciding whether to build VectorTestBase.* files (velox_vector_test_lib).

This is not a problem with velox_exec_test_util that contains OperatorTestBase.* because that is being build even if VELOX_BUILD_TESTING is off, but VELOX_BUILD_TEST_UTILS is on.

presto-native-execution/velox/velox/exec/CMakeLists.txt

if(${VELOX_BUILD_TESTING}) add_subdirectory(tests) add_subdirectory(benchmarks) elseif(${VELOX_BUILD_TEST_UTILS}) add_subdirectory(tests/utils) endif()

When building Prestissimo, we set VELOX_BUILD_TESTING OFF and VELOX_BUILD_TEST_UTILS ON.

presto-native-execution/CMakeLists.txt

set(VELOX_BUILD_TESTING OFF CACHE BOOL "Enable Velox tests") option(PRESTO_ENABLE_TESTING "Enable tests" ON) set(VELOX_BUILD_TEST_UTILS ${PRESTO_ENABLE_TESTING} CACHE BOOL "Enable Velox test utils")

A fix could be to move to call add_subdirectory(tests) unconditionally in vector/CMakeLists.txt, then build velox_vector_test_lib if VELOX_BUILD_TEST_UTILS is ON and the rest of the tests if VELOX_BUILD_TESTING is ON. Alternatively, we could move VectorMaker.* and VectorTestBase.* into a separate directory and call add_directory for that if VELOX_BUILD_TEST_UTILS is ON.

CC: @majetideepak @kgpai

@mbasmanova I tried the solution #1 and it worked! So I'm using vector maker and also assertEqualityVectors ... in this PR as well. Preparing a Velox PR for that change too.

This method is not needed. Please, remove.

mbasmanova · 2022-09-12T20:35:42Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+    return result;
+  }
+
+  std::shared_ptr<const RowType> rowType(


This method and a few following methods appear not used. Would you double check?

Actually only this method.

This method still needs to be deleted.

mbasmanova · 2022-09-13T01:09:07Z

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.cpp

+    auto numPartitions = planNode->numPartitions();
+    VELOX_CHECK(!planNode->keys().empty(), "Empty keys for hive hash!");
+    std::vector<int> bucketToPartition(numPartitions);
+    for (int i = 0; i < numPartitions; ++i) {


nit: consider using std::iota for brevity

std::iota(bucketToPartition.begin(), bucketToPartition.end(), 0);

mbasmanova · 2022-09-13T01:10:05Z

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.cpp

+#include "presto_cpp/main/operators/PartitionAndSerialize.h"
+#include "velox/connectors/hive/HivePartitionFunction.h"
+#include "velox/exec/HashPartitionFunction.h"
+#include "velox/expression/Expr.h"


Is this #include needed?

I'll remove it.

mbasmanova · 2022-09-13T16:54:23Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+        std::vector<VectorPtr>{});
+  }
+
+  RowVectorPtr makeRowVector(


This method should not be needed. We should use the implementation from the base class. Then, 'rowVector' method won't be needed as well.

mbasmanova

@beroyfb Behnam, thank you for iterating on this PR. Some comments for the first commit.

mbasmanova · 2022-09-20T20:32:02Z

presto-native-execution/presto_cpp/main/operators/CMakeLists.txt

+        velox_hive_partition_function
+        velox_vector)
+
+add_subdirectory(tests)


Wrap this in if(PRESTO_ENABLE_TESTING)

mbasmanova · 2022-09-20T20:33:38Z

presto-native-execution/presto_cpp/main/operators/PartitionAndSerialize.h

+        velox::ROW(
+            {"partition", "data"}, {velox::INTEGER(), velox::VARBINARY()})
+            ->equivalent(*outputType_));
+    VELOX_CHECK(!keys_.empty(), "Empty keys for hive hash!");


VELOX_USER_CHECK + remove ! from the message (no need to scream at the user)

mbasmanova · 2022-09-20T20:34:18Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+}
+} // namespace
+
+class UnsaferowShuffleTest : public exec::test::OperatorTestBase {


Still needs to be fixed.

mbasmanova · 2022-09-20T20:35:01Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+    return result;
+  }
+
+  std::shared_ptr<const RowType> rowType(


This method still needs to be deleted.

mbasmanova · 2022-09-20T20:35:22Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+    return ROW(std::move(names), std::move(types));
+  }
+
+  RowVectorPtr rowVector(const std::vector<VectorPtr>& children) {


This method should not be needed. It is available in the base class.

mbasmanova · 2022-09-20T20:35:40Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+        std::vector<VectorPtr>{});
+  }
+
+  RowVectorPtr makeRowVector(


mbasmanova · 2022-09-20T20:35:58Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+    return rowVec;
+  }
+
+  void assertEqualVectors(const VectorPtr& expected, const VectorPtr& actual) {


This method is not needed. Please, remove.

mbasmanova · 2022-09-20T20:36:31Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+  }
+}
+
+TEST_F(UnsafeRowShuffleTest, partitionAndSerializeStringTest) {


naming: partitionAndSerializeStringTest -> partitionAndSerializeToString

mbasmanova · 2022-09-20T20:36:40Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+}
+
+TEST_F(UnsafeRowShuffleTest, partitionAndSerializeStringTest) {
+  exec::Operator::registerOperator(


This is not needed

mbasmanova · 2022-09-20T20:37:35Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

+          .planNode();
+
+  ASSERT_EQ(
+      plan->toString(true, false),


Would you also add a test case with toString(false, false) and toString(true, true)?

mbasmanova

2nd commit looks good % one request.

mbasmanova · 2022-09-20T20:38:52Z

presto-native-execution/presto_cpp/main/operators/ShuffleInterface.h

+  virtual void collect(int32_t partition, std::string_view data) = 0;
+
+  // Tell the shuffle system the writer is done.
+  // The input parameter success set to false indicate aborted client.


mbasmanova · 2022-09-20T20:39:20Z

presto-native-execution/presto_cpp/main/operators/ShuffleInterface.h

+  virtual bool hasNext(int32_t partition) const = 0;
+
+  // Read the next block of data for this partition.
+  virtual velox::BufferPtr next(int32_t partition) = 0;


Do we need a method to tell the reader we are done reading (either successfully or not)?

mbasmanova · 2022-09-20T20:40:35Z

presto-native-execution/presto_cpp/main/operators/ShuffleWrite.h

+
+namespace facebook::presto::operators {
+
+class ShuffleWriteNode : public velox::core::PlanNode {


Would you add a test for ShuffleWriteNode->toString()?

mbasmanova

Remaining comments for the 3rd commit.

mbasmanova · 2022-09-20T20:41:23Z

.github/workflows/test-native.yml

@@ -92,4 +92,5 @@ jobs:
          ./mvnw install ${MAVEN_FAST_INSTALL} -pl '!presto-docs,!presto-server,!presto-server-rpm'

      - name: Run e2e tests
+        rm -rf /tmp/hive_data/tpch/


Is this needed? If so, can you extract this into a separate commit ?

mbasmanova · 2022-09-20T20:42:13Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

@@ -212,83 +232,17 @@ class UnsaferowShuffleTest : public exec::test::OperatorTestBase {
    serializer->deserialize(input.get(), pool(), rowType, &result, nullptr);
    return result;
  }
-
-  std::shared_ptr<const RowType> rowType(


Would you move these deletes to the first commit to avoid code churn?

mbasmanova · 2022-09-20T20:42:54Z

presto-native-execution/presto_cpp/main/operators/tests/UnsaferowShuffleTest.cpp

  exec::Operator::registerOperator(
      std::make_unique<PartitionAndSerializeTranslator>());

-  auto data = makeRowVector({
+  auto data = vectorMaker_.rowVector({


It would be possible to use makeRowVector here. Is this not working?

mbasmanova

@beroyfb Looks good to me. Thank you for working on adding shuffle support to Sapphire on Velox.

PartitionAndSerialize partitions and also serializes rows in the row vector as a vector of string (each string holding a serialized row). It uses hive partition function. Later we may add murmur3 hash as well.

ShuffleWrite writes the result of serialized vectors into an external shuffle system. Also adding external shuffle interface.

Also an end2end test simulating an external shuffle service for unsaferow blocks.

beroyfb requested a review from a team as a code owner September 12, 2022 05:21

beroyfb force-pushed the sapphire-shuffle-operators branch 2 times, most recently from d94ec59 to 4d81db3 Compare September 12, 2022 18:08

mbasmanova reviewed Sep 12, 2022

View reviewed changes

beroyfb force-pushed the sapphire-shuffle-operators branch from 4d81db3 to a9c03ae Compare September 12, 2022 23:56

mbasmanova reviewed Sep 13, 2022

View reviewed changes

beroyfb force-pushed the sapphire-shuffle-operators branch 2 times, most recently from 32efc6a to 387ca07 Compare September 13, 2022 16:52

mbasmanova reviewed Sep 13, 2022

View reviewed changes

beroyfb mentioned this pull request Sep 13, 2022

Build velox_vector_test_lib if VELOX_BUILD_TEST_UTILS is ON facebookincubator/velox#2531

Closed

beroyfb force-pushed the sapphire-shuffle-operators branch 3 times, most recently from ee9dad1 to 349a638 Compare September 14, 2022 06:10

mbasmanova changed the title ~~Operators for unsaferow shuffle~~ [native] Operators for unsaferow shuffle Sep 14, 2022

beroyfb force-pushed the sapphire-shuffle-operators branch 4 times, most recently from 25396cc to 49299e0 Compare September 20, 2022 18:10

beroyfb requested a review from a team as a code owner September 20, 2022 18:10

beroyfb requested a review from ChunxuTang September 20, 2022 18:10

mbasmanova reviewed Sep 20, 2022

View reviewed changes

beroyfb force-pushed the sapphire-shuffle-operators branch from 49299e0 to 8fd55a8 Compare September 21, 2022 01:41

beroyfb force-pushed the sapphire-shuffle-operators branch from 8fd55a8 to 6068e73 Compare September 21, 2022 06:17

mbasmanova approved these changes Sep 21, 2022

View reviewed changes

beroyfb added 3 commits September 21, 2022 07:15

Add new sapphire operator PartitionAndSerialize

eeb73d8

PartitionAndSerialize partitions and also serializes rows in the row vector as a vector of string (each string holding a serialized row). It uses hive partition function. Later we may add murmur3 hash as well.

Add ShuffleWrite operator for external shuffle support

4cda121

ShuffleWrite writes the result of serialized vectors into an external shuffle system. Also adding external shuffle interface.

Add Exchange source for unsaferow blocks

6c51998

Also an end2end test simulating an external shuffle service for unsaferow blocks.

beroyfb force-pushed the sapphire-shuffle-operators branch from 6068e73 to 6c51998 Compare September 21, 2022 14:16

spershin merged commit a6e84c0 into prestodb:master Sep 21, 2022


		namespace facebook::presto::operators {

		class UnsafeRowExchangeSource : public velox::exec::ExchangeSource {


		namespace facebook::presto::operators {

		class ShuffleWriteNode : public velox::core::PlanNode {

[native] Operators for unsaferow shuffle #18327

[native] Operators for unsaferow shuffle #18327

Conversation

beroyfb commented Sep 12, 2022 • edited Loading

linux-foundation-easycla bot commented Sep 12, 2022 • edited Loading

mbasmanova left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mbasmanova left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mbasmanova left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mbasmanova left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mbasmanova left a comment

Choose a reason for hiding this comment

beroyfb commented Sep 12, 2022 •

edited

Loading

linux-foundation-easycla bot commented Sep 12, 2022 •

edited

Loading