feat(init): add FLINK preset with Flink-flavored function registration

src/main/cpp/CMakeLists.txt

@@ -75,14 +75,14 @@ file(STRINGS ${VELOX_REF_HASH_FILE} VELOX_SOURCE_URL_MD5)
 if(VELOX4J_VELOX_SOURCE_DIR)
   cpmaddpackage(NAME velox SOURCE_DIR ${VELOX4J_VELOX_SOURCE_DIR})
 else()
-  set(VELOX_SOURCE_URL "https://github.com/bigo-sg/velox.git")
+  set(VELOX_SOURCE_URL "https://github.com/ggjh-159/velox.git")
   cpmaddpackage(
     NAME
     velox
     GIT_REPOSITORY
     ${VELOX_SOURCE_URL}
     GIT_TAG
-    ${VELOX_REF})
+    feat/functions-flinksql)
 endif()
 if(TARGET gflags::gflags_shared)
   get_target_property(GFLAGS_SHARED_LIBRARY gflags::gflags_shared

src/main/cpp/main/velox4j/init/Config.cc

@@ -19,7 +19,35 @@
 
 namespace velox4j {
 using namespace facebook::velox;
+
+namespace {
+std::string presetToStr(Preset p) {
+  switch (p) {
+    case Preset::SPARK:
+      return "SPARK";
+    case Preset::FLINK:
+      return "FLINK";
+  }
+  VELOX_FAIL("Unknown Preset value: {}", static_cast<int>(p));
+}
+
+Preset presetFromString(const std::string& key, const std::string& value) {
+  if (value == "SPARK") {
+    return Preset::SPARK;
+  }
+  if (value == "FLINK") {
+    return Preset::FLINK;
+  }
+  VELOX_FAIL(
+      "Invalid configuration for key '{}'. Value '{}' cannot be converted to type velox4j::Preset (expected SPARK or FLINK).",
+      key,
+      value);
+}
+} // namespace
+
 config::ConfigBase::Entry<Preset> VELOX4J_INIT_PRESET(
     "velox4j.init.preset",
-    Preset::SPARK);
+    Preset::SPARK,
+    presetToStr,
+    presetFromString);
 } // namespace velox4j

src/main/cpp/main/velox4j/init/Config.h

@@ -20,7 +20,7 @@
 #include <velox/common/config/Config.h>
 
 namespace velox4j {
-enum Preset { SPARK = 0 };
+enum Preset { SPARK = 0, FLINK = 1 };
 
 extern facebook::velox::config::ConfigBase::Entry<Preset> VELOX4J_INIT_PRESET;
 } // namespace velox4j

src/main/cpp/main/velox4j/init/Init.cc

@@ -55,6 +55,7 @@
 #include <velox/functions/sparksql/aggregates/Register.h>
 #include <velox/functions/sparksql/registration/Register.h>
 #include <velox/functions/sparksql/window/WindowFunctionsRegistration.h>
+#include <velox/functions/flinksql/Register.h>
 #include <velox/connectors/fuzzer/DiscardDataSink.cpp>
 #include <velox/connectors/fuzzer/FuzzerConnector.cpp>
 #include <velox/vector/fuzzer/ConstrainedVectorGenerator.cpp>
@@ -81,53 +82,18 @@ void init(const std::function<void()>& f) {
   f();
 }
 
-void initForSpark() {
-  FLAGS_velox_memory_leak_check_enabled = true;
-  FLAGS_velox_memory_pool_capacity_transfer_across_tasks = true;
-  FLAGS_velox_exception_user_stacktrace_enabled = true;
-  FLAGS_velox_exception_system_stacktrace_enabled = true;
-  filesystems::registerLocalFileSystem();
-  filesystems::registerHdfsFileSystem();
-  memory::MemoryManager::initialize({});
-  dwio::common::registerFileSinks();
-  dwrf::registerDwrfReaderFactory();
-  dwrf::registerDwrfWriterFactory();
-  dwrf::registerOrcWriterFactory();
-  parquet::registerParquetReaderFactory();
-  parquet::registerParquetWriterFactory();
-  text::registerTextWriterFactory();
-  functions::sparksql::registerFunctions();
-
-  // Override to return NULL on no match for Flink compatibility.
-  exec::registerStatefulVectorFunction(
-      "regexp_extract",
-      functions::re2ExtractSignatures(),
-      [](const std::string& name,
-         const std::vector<exec::VectorFunctionArg>& inputArgs,
-         const core::QueryConfig& config) {
-        return functions::makeRe2Extract(
-            name, inputArgs, config, /*emptyNoMatch=*/false);
-      });
-  aggregate::prestosql::registerAllAggregateFunctions(
-      "",
-      true /*registerCompanionFunctions*/,
-      false /*onlyPrestoSignatures*/,
-      true /*overwrite*/);
-  functions::aggregate::sparksql::registerAggregateFunctions(
-      "", true /*registerCompanionFunctions*/, true /*overwrite*/);
-  window::prestosql::registerAllWindowFunctions();
-  functions::window::sparksql::registerWindowFunctions("");
-  stateful::udf::registerFunctions();
-
-  ConfigArray::registerSerDe();
-  ConnectorConfigArray::registerSerDe();
-  Evaluation::registerSerDe();
-  Query::registerSerDe();
-  Type::registerSerDe();
-  common::Filter::registerSerDe();
+void registerConnectors() {
+  // fuzzer
   connector::fuzzer::DiscardDataTableHandle::registerSerDe();
   connector::fuzzer::FuzzerTableHandle::registerSerDe();
   connector::fuzzer::FuzzerConnectorSplit::registerSerDe();
+  connector::registerConnector(
+      std::make_shared<connector::fuzzer::FuzzerConnector>(
+          "connector-fuzzer",
+          std::make_shared<facebook::velox::config::ConfigBase>(
+              std::unordered_map<std::string, std::string>()),
+          nullptr));
+  // hive
   connector::hive::HiveTableHandle::registerSerDe();
   connector::hive::LocationHandle::registerSerDe();
   connector::hive::HiveColumnHandle::registerSerDe();
@@ -143,6 +109,7 @@ void initForSpark() {
       std::make_shared<facebook::velox::config::ConfigBase>(
           std::unordered_map<std::string, std::string>()),
       nullptr));
+  // kafka
   connector::kafka::KafkaTableHandle::registerSerDe();
   connector::kafka::KafkaConnectorSplit::registerSerDe();
   connector::registerConnector(
@@ -151,6 +118,7 @@ void initForSpark() {
           std::make_shared<facebook::velox::config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           nullptr));
+  // pulsar
   connector::pulsar::PulsarTableHandle::registerSerDe();
   connector::pulsar::PulsarConnectorSplit::registerSerDe();
   connector::registerConnector(
@@ -159,25 +127,22 @@ void initForSpark() {
           std::make_shared<facebook::velox::config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           nullptr));
+  // filesystem
   connector::filesystem::FileSystemInsertTableHandle::registerSerDe();
   connector::registerConnector(
       std::make_shared<connector::filesystem::FileSystemConnector>(
           "connector-filesystem",
           std::make_shared<facebook::velox::config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           nullptr));
+  // external-stream
   ExternalStreamConnectorSplit::registerSerDe();
   ExternalStreamTableHandle::registerSerDe();
   connector::registerConnector(std::make_shared<ExternalStreamConnector>(
       "connector-external-stream",
       std::make_shared<facebook::velox::config::ConfigBase>(
           std::unordered_map<std::string, std::string>())));
-  connector::registerConnector(
-      std::make_shared<connector::fuzzer::FuzzerConnector>(
-          "connector-fuzzer",
-          std::make_shared<facebook::velox::config::ConfigBase>(
-              std::unordered_map<std::string, std::string>()),
-          nullptr));
+  // nexmark
   connector::nexmark::NexmarkTableHandle::registerSerDe();
   connector::nexmark::NexmarkConnectorSplit::registerSerDe();
   connector::registerConnector(
@@ -186,13 +151,15 @@ void initForSpark() {
           std::make_shared<facebook::velox::config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           nullptr));
+  // print
   connector::print::PrintTableHandle::registerSerDe();
   connector::registerConnector(
       std::make_shared<connector::print::PrintConnector>(
           "connector-print",
           std::make_shared<facebook::velox::config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           nullptr));
+  // from-elements
   connector::from_elements::FromElementsTableHandle::registerSerDe();
   connector::from_elements::FromElementsConnectorSplit::registerSerDe();
   connector::registerConnector(
@@ -201,6 +168,45 @@ void initForSpark() {
           std::make_shared<facebook::velox::config::ConfigBase>(
               std::unordered_map<std::string, std::string>()),
           nullptr));
+}
+
+void initForSpark() {
+  FLAGS_velox_memory_leak_check_enabled = true;
+  FLAGS_velox_memory_pool_capacity_transfer_across_tasks = true;
+  FLAGS_velox_exception_user_stacktrace_enabled = true;
+  FLAGS_velox_exception_system_stacktrace_enabled = true;
+  filesystems::registerLocalFileSystem();
+  filesystems::registerHdfsFileSystem();
+  memory::MemoryManager::initialize({});
+  dwio::common::registerFileSinks();
+  dwrf::registerDwrfReaderFactory();
+  dwrf::registerDwrfWriterFactory();
+  dwrf::registerOrcWriterFactory();
+  parquet::registerParquetReaderFactory();
+  parquet::registerParquetWriterFactory();
+  text::registerTextWriterFactory();
+  functions::sparksql::registerFunctions();
+
+  aggregate::prestosql::registerAllAggregateFunctions(
+      "",
+      true /*registerCompanionFunctions*/,
+      false /*onlyPrestoSignatures*/,
+      true /*overwrite*/);
+  functions::aggregate::sparksql::registerAggregateFunctions(
+      "", true /*registerCompanionFunctions*/, true /*overwrite*/);
+  window::prestosql::registerAllWindowFunctions();
+  functions::window::sparksql::registerWindowFunctions("");
+  stateful::udf::registerFunctions();
+
+  ConfigArray::registerSerDe();
+  ConnectorConfigArray::registerSerDe();
+  Evaluation::registerSerDe();
+  Query::registerSerDe();
+  Type::registerSerDe();
+  common::Filter::registerSerDe();
+
+  registerConnectors();
+
   core::PlanNode::registerSerDe();
   stateful::StatefulPlanNode::registerSerDe();
   stateful::KeyedStateBackendParameters::registerSerDe();
@@ -209,6 +215,18 @@ void initForSpark() {
   core::ITypedExpr::registerSerDe();
   exec::registerPartitionFunctionSerDe();
 }
+
+void initForFlink() {
+  // FLINK preset reuses initForSpark() because velox/functions/flinksql/ does
+  // not yet provide a complete scalar/aggregate/window function set, so Flink
+  // still relies on sparksql's registration as the baseline. The flinksql
+  // overrides below layer on top to fix dialect-specific semantics (e.g.
+  // REGEXP_EXTRACT returns NULL on no match instead of empty string).
+  // TODO: complete velox/functions/flinksql/ registration so Flink no longer
+  // depends on sparksql.
+  initForSpark();
+  functions::flinksql::registerFunctions();
+}
 } // namespace
 
 void initialize(const std::shared_ptr<ConfigArray>& configArray) {
@@ -222,6 +240,9 @@ void initialize(const std::shared_ptr<ConfigArray>& configArray) {
       case SPARK:
         initForSpark();
         break;
+      case FLINK:
+        initForFlink();
+        break;
       default:
         VELOX_FAIL("Unknown preset: {}", folly::to<std::string>(preset));
     }

src/main/cpp/test/velox4j/test/Init.h

@@ -27,4 +27,12 @@ inline void testingEnsureInitializedForSpark() {
           {VELOX4J_INIT_PRESET.key, folly::to<std::string>(Preset::SPARK)}});
   std::call_once(flag, [&]() { initialize(conf); });
 }
+
+inline void testingEnsureInitializedForFlink() {
+  static std::once_flag flag;
+  auto conf = std::make_shared<ConfigArray>(
+      std::vector<std::pair<std::string, std::string>>{
+          {VELOX4J_INIT_PRESET.key, folly::to<std::string>(Preset::FLINK)}});
+  std::call_once(flag, [&]() { initialize(conf); });
+}
 } // namespace velox4j

src/main/java/io/github/zhztheplayer/velox4j/config/Preset.java

@@ -0,0 +1,25 @@
+/*
+* 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 io.github.zhztheplayer.velox4j.config;
+
+public final class Preset {
+  public static final String KEY = "velox4j.init.preset";
+  public static final String SPARK = "SPARK";
+  public static final String FLINK = "FLINK";
+
+  private Preset() {}
+}

src/test/java/io/github/zhztheplayer/velox4j/functions/RegexpExtractTest.java

@@ -59,7 +59,7 @@ public class RegexpExtractTest {
 
   @BeforeClass
   public static void beforeClass() throws Exception {
-    Velox4jTests.ensureInitialized();
+    Velox4jTests.ensureInitializedForFlink();
     memoryManager = MemoryManager.create(AllocationListener.NOOP);
   }
 

src/test/java/io/github/zhztheplayer/velox4j/test/Velox4jTests.java

@@ -19,6 +19,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.github.zhztheplayer.velox4j.Velox4j;
+import io.github.zhztheplayer.velox4j.config.Preset;
 
 public class Velox4jTests {
   private static final AtomicBoolean initialized = new AtomicBoolean(false);
@@ -29,4 +30,12 @@ public static void ensureInitialized() {
     }
     Velox4j.initialize();
   }
+
+  public static void ensureInitializedForFlink() {
+    if (!initialized.compareAndSet(false, true)) {
+      return;
+    }
+    Velox4j.configure(Preset.KEY, Preset.FLINK);
+    Velox4j.initialize();
+  }
 }