From 98c2aff93f60399b19c053d1f52f12922fd91d45 Mon Sep 17 00:00:00 2001 From: ggjh-159 Date: Mon, 6 Jul 2026 11:11:02 +0800 Subject: [PATCH] feat(init): add FLINK preset with Flink-flavored function registration --- src/main/cpp/CMakeLists.txt | 4 +- src/main/cpp/main/velox4j/init/Config.cc | 30 ++++- src/main/cpp/main/velox4j/init/Config.h | 2 +- src/main/cpp/main/velox4j/init/Init.cc | 121 ++++++++++-------- src/main/cpp/test/velox4j/test/Init.h | 8 ++ .../zhztheplayer/velox4j/config/Preset.java | 25 ++++ .../velox4j/functions/RegexpExtractTest.java | 2 +- .../velox4j/test/Velox4jTests.java | 9 ++ 8 files changed, 146 insertions(+), 55 deletions(-) create mode 100644 src/main/java/io/github/zhztheplayer/velox4j/config/Preset.java diff --git a/src/main/cpp/CMakeLists.txt b/src/main/cpp/CMakeLists.txt index 53b14eb3b1..fb13d12b22 100644 --- a/src/main/cpp/CMakeLists.txt +++ b/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 diff --git a/src/main/cpp/main/velox4j/init/Config.cc b/src/main/cpp/main/velox4j/init/Config.cc index a37d86eee2..2d69a34a52 100644 --- a/src/main/cpp/main/velox4j/init/Config.cc +++ b/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(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 VELOX4J_INIT_PRESET( "velox4j.init.preset", - Preset::SPARK); + Preset::SPARK, + presetToStr, + presetFromString); } // namespace velox4j diff --git a/src/main/cpp/main/velox4j/init/Config.h b/src/main/cpp/main/velox4j/init/Config.h index 2eb996b927..66fc1351e7 100644 --- a/src/main/cpp/main/velox4j/init/Config.h +++ b/src/main/cpp/main/velox4j/init/Config.h @@ -20,7 +20,7 @@ #include namespace velox4j { -enum Preset { SPARK = 0 }; +enum Preset { SPARK = 0, FLINK = 1 }; extern facebook::velox::config::ConfigBase::Entry VELOX4J_INIT_PRESET; } // namespace velox4j diff --git a/src/main/cpp/main/velox4j/init/Init.cc b/src/main/cpp/main/velox4j/init/Init.cc index fd5682d961..cf6908cba5 100644 --- a/src/main/cpp/main/velox4j/init/Init.cc +++ b/src/main/cpp/main/velox4j/init/Init.cc @@ -55,6 +55,7 @@ #include #include #include +#include #include #include #include @@ -81,53 +82,18 @@ void init(const std::function& 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& 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", + std::make_shared( + std::unordered_map()), + nullptr)); + // hive connector::hive::HiveTableHandle::registerSerDe(); connector::hive::LocationHandle::registerSerDe(); connector::hive::HiveColumnHandle::registerSerDe(); @@ -143,6 +109,7 @@ void initForSpark() { std::make_shared( std::unordered_map()), nullptr)); + // kafka connector::kafka::KafkaTableHandle::registerSerDe(); connector::kafka::KafkaConnectorSplit::registerSerDe(); connector::registerConnector( @@ -151,6 +118,7 @@ void initForSpark() { std::make_shared( std::unordered_map()), nullptr)); + // pulsar connector::pulsar::PulsarTableHandle::registerSerDe(); connector::pulsar::PulsarConnectorSplit::registerSerDe(); connector::registerConnector( @@ -159,6 +127,7 @@ void initForSpark() { std::make_shared( std::unordered_map()), nullptr)); + // filesystem connector::filesystem::FileSystemInsertTableHandle::registerSerDe(); connector::registerConnector( std::make_shared( @@ -166,18 +135,14 @@ void initForSpark() { std::make_shared( std::unordered_map()), nullptr)); + // external-stream ExternalStreamConnectorSplit::registerSerDe(); ExternalStreamTableHandle::registerSerDe(); connector::registerConnector(std::make_shared( "connector-external-stream", std::make_shared( std::unordered_map()))); - connector::registerConnector( - std::make_shared( - "connector-fuzzer", - std::make_shared( - std::unordered_map()), - nullptr)); + // nexmark connector::nexmark::NexmarkTableHandle::registerSerDe(); connector::nexmark::NexmarkConnectorSplit::registerSerDe(); connector::registerConnector( @@ -186,6 +151,7 @@ void initForSpark() { std::make_shared( std::unordered_map()), nullptr)); + // print connector::print::PrintTableHandle::registerSerDe(); connector::registerConnector( std::make_shared( @@ -193,6 +159,7 @@ void initForSpark() { std::make_shared( std::unordered_map()), nullptr)); + // from-elements connector::from_elements::FromElementsTableHandle::registerSerDe(); connector::from_elements::FromElementsConnectorSplit::registerSerDe(); connector::registerConnector( @@ -201,6 +168,45 @@ void initForSpark() { std::make_shared( std::unordered_map()), 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) { @@ -222,6 +240,9 @@ void initialize(const std::shared_ptr& configArray) { case SPARK: initForSpark(); break; + case FLINK: + initForFlink(); + break; default: VELOX_FAIL("Unknown preset: {}", folly::to(preset)); } diff --git a/src/main/cpp/test/velox4j/test/Init.h b/src/main/cpp/test/velox4j/test/Init.h index dfe1b486fe..e6209791c4 100644 --- a/src/main/cpp/test/velox4j/test/Init.h +++ b/src/main/cpp/test/velox4j/test/Init.h @@ -27,4 +27,12 @@ inline void testingEnsureInitializedForSpark() { {VELOX4J_INIT_PRESET.key, folly::to(Preset::SPARK)}}); std::call_once(flag, [&]() { initialize(conf); }); } + +inline void testingEnsureInitializedForFlink() { + static std::once_flag flag; + auto conf = std::make_shared( + std::vector>{ + {VELOX4J_INIT_PRESET.key, folly::to(Preset::FLINK)}}); + std::call_once(flag, [&]() { initialize(conf); }); +} } // namespace velox4j diff --git a/src/main/java/io/github/zhztheplayer/velox4j/config/Preset.java b/src/main/java/io/github/zhztheplayer/velox4j/config/Preset.java new file mode 100644 index 0000000000..b18d9bd195 --- /dev/null +++ b/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() {} +} diff --git a/src/test/java/io/github/zhztheplayer/velox4j/functions/RegexpExtractTest.java b/src/test/java/io/github/zhztheplayer/velox4j/functions/RegexpExtractTest.java index c30cef238a..3776f680ac 100644 --- a/src/test/java/io/github/zhztheplayer/velox4j/functions/RegexpExtractTest.java +++ b/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); } diff --git a/src/test/java/io/github/zhztheplayer/velox4j/test/Velox4jTests.java b/src/test/java/io/github/zhztheplayer/velox4j/test/Velox4jTests.java index d64ae8fa24..3ed96c5185 100644 --- a/src/test/java/io/github/zhztheplayer/velox4j/test/Velox4jTests.java +++ b/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(); + } }