diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4c3378bf..d0a69d88 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,16 +101,14 @@ jobs: run: | bin/build-release.sh --host-only version="$(mvn -q -DforceStdout help:evaluate -Dexpression=project.version)" - jar tf "streamfusion-kafka/target/streamfusion-kafka-${version}.jar" | grep -q 'native/kafka/linux/x86_64/libstreamfusion_kafka.so' - jar tf "streamfusion-json/target/streamfusion-json-${version}.jar" | grep -q 'native/json/linux/x86_64/libstreamfusion_json.so' - jar tf "streamfusion-csv/target/streamfusion-csv-${version}.jar" | grep -q 'native/csv/linux/x86_64/libstreamfusion_csv.so' - jar tf "streamfusion-raw/target/streamfusion-raw-${version}.jar" | grep -q 'native/raw/linux/x86_64/libstreamfusion_raw.so' - jar tf "streamfusion-avro/target/streamfusion-avro-${version}.jar" | grep -q 'native/avro/linux/x86_64/libstreamfusion_avro.so' - if jar tf "streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-${version}.jar" | grep -q 'libstreamfusion_avro'; then + line="$(mvn -q -DforceStdout help:evaluate -Dexpression=flink.line)" + for module in kafka json csv raw avro protobuf parquet; do + jar tf "streamfusion-${module}/target/streamfusion-${module}-flink${line}-${version}.jar" \ + | grep -q "native/${module}/linux/x86_64/libstreamfusion_${module}.so" + done + if jar tf "streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-flink${line}-${version}.jar" | grep -q 'libstreamfusion_avro'; then exit 1 fi - jar tf "streamfusion-protobuf/target/streamfusion-protobuf-${version}.jar" | grep -q 'native/protobuf/linux/x86_64/libstreamfusion_protobuf.so' - jar tf "streamfusion-parquet/target/streamfusion-parquet-${version}.jar" | grep -q 'native/parquet/linux/x86_64/libstreamfusion_parquet.so' bin/check-artifacts.sh --host-only bin/build-flink-image.sh --tag streamfusion-flink:image-it --load --skip-release-build diff --git a/bin/build-flink-image.sh b/bin/build-flink-image.sh index 3be768c2..a062c57c 100755 --- a/bin/build-flink-image.sh +++ b/bin/build-flink-image.sh @@ -13,6 +13,7 @@ Builds a job-neutral StreamFusion Flink base image. --load Build one platform and load it into the local Docker daemon. --platform Platform for --load (default: Docker server platform). --flink-image Flink base image (default: flink:2.2.1-scala_2.12-java17). + --flink-line StreamFusion Flink line to build (default: 2.2). Must match --flink-image. --skip-release-build Reuse the already-built StreamFusion JARs. EOF exit 64 @@ -21,6 +22,7 @@ EOF script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) repo_root=$(cd "$script_dir/.." && pwd) flink_image=flink:2.2.1-scala_2.12-java17 +flink_line=2.2 image_tag= mode= platform= @@ -48,6 +50,11 @@ while [ "$#" -gt 0 ]; do flink_image=$2 shift 2 ;; + --flink-line) + [ "$#" -ge 2 ] || usage + flink_line=$2 + shift 2 + ;; --skip-release-build) skip_release_build=true shift @@ -67,12 +74,12 @@ command -v docker >/dev/null 2>&1 || { docker buildx version >/dev/null if [ "$skip_release_build" = false ]; then - "$repo_root/bin/build-release.sh" --linux-only + "$repo_root/bin/build-release.sh" --linux-only --flink-line "$flink_line" fi artifact_version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version) -loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-$artifact_version.jar -core_jar=$repo_root/streamfusion-core/target/streamfusion-core-$artifact_version-runtime.jar +loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$artifact_version.jar +core_jar=$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$artifact_version-runtime.jar [ -f "$loader_jar" ] && [ -f "$core_jar" ] || { echo "StreamFusion release JARs are missing; run bin/build-release.sh first." >&2 exit 66 @@ -104,6 +111,7 @@ docker buildx build \ --platform "$platforms" \ --build-arg "FLINK_IMAGE=$flink_image" \ --build-arg "STREAMFUSION_VERSION=$artifact_version" \ + --build-arg "STREAMFUSION_FLINK_LINE=$flink_line" \ --tag "$image_tag" \ --file "$repo_root/docker/flink-base.Dockerfile" \ "$output" \ diff --git a/bin/build-release.sh b/bin/build-release.sh index c62a4c3c..57e90dc6 100755 --- a/bin/build-release.sh +++ b/bin/build-release.sh @@ -2,10 +2,10 @@ set -eu -if [ "$#" -gt 1 ] || { [ "$#" -eq 1 ] && [ "$1" != "--host-only" ] && [ "$1" != "--linux-only" ]; }; then - echo "usage: $0 [--host-only | --linux-only]" >&2 +usage() { + echo "usage: $0 [--host-only | --linux-only] [--flink-line ]" >&2 exit 64 -fi +} script_dir=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd) repo_root=$(cd "$script_dir/.." && pwd) @@ -13,13 +13,33 @@ native_dir=$repo_root/native stage_dir=$native_dir/target/universal host_only=false linux_only=false +# Deployable coordinates carry the Flink line, so a release targets one line at a time. +flink_line=2.2 + +while [ "$#" -gt 0 ]; do + case $1 in + --host-only) host_only=true ;; + --linux-only) linux_only=true ;; + --flink-line) + [ "$#" -ge 2 ] || usage + flink_line=$2 + shift + ;; + *) usage ;; + esac + shift +done -if [ "$#" -eq 1 ] && [ "$1" = "--host-only" ]; then - host_only=true -elif [ "$#" -eq 1 ]; then - linux_only=true +if [ "$host_only" = true ] && [ "$linux_only" = true ]; then + usage fi +case $flink_line in + 2.2) flink_profile= ;; + 2.1) flink_profile=,flink-2.1 ;; + *) echo "unsupported Flink line: $flink_line" >&2; exit 64 ;; +esac + host_platform() { case "$(uname -s)" in Linux) printf '%s\n' linux ;; @@ -176,4 +196,4 @@ fi # platform build. A release always starts from empty Java output directories. The release profile # builds the same source and javadoc attachments as the publish workflow, unsigned, so attachment # failures surface here instead of on the release runner. -(cd "$repo_root" && mvn clean package -Pdist,universal,release -Dgpg.skip=true -DskipTests) +(cd "$repo_root" && mvn clean package -Pdist,universal,release${flink_profile} -Dgpg.skip=true -DskipTests) diff --git a/bin/check-artifacts.sh b/bin/check-artifacts.sh index 45ef83f7..fe22b384 100755 --- a/bin/check-artifacts.sh +++ b/bin/check-artifacts.sh @@ -25,6 +25,8 @@ fi script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) repo_root=$(cd "$script_dir/.." && pwd) version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version) +# Module directories are line-neutral; the artifacts they build are not. +flink_line=${FLINK_LINE:-2.2} modules="core kafka json csv raw avro avro-confluent-registry protobuf parquet" entries=$(mktemp) native_entries=$(mktemp) @@ -69,10 +71,11 @@ assert_no_native_payload() { for suffix in $modules; do module="streamfusion-$suffix" + artifact="$module-flink$flink_line" if [ "$suffix" = core ]; then - jar_file="$repo_root/$module/target/$module-$version-runtime.jar" + jar_file="$repo_root/$module/target/$artifact-$version-runtime.jar" else - jar_file="$repo_root/$module/target/$module-$version.jar" + jar_file="$repo_root/$module/target/$artifact-$version.jar" fi if [ ! -f "$jar_file" ]; then echo "missing artifact: $jar_file" >&2 @@ -87,8 +90,8 @@ for suffix in $modules; do fi done -core_jar="$repo_root/streamfusion-core/target/streamfusion-core-$version-runtime.jar" -core_main_jar="$repo_root/streamfusion-core/target/streamfusion-core-$version.jar" +core_jar="$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$version-runtime.jar" +core_main_jar="$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$version.jar" assert_native_payload "$core_main_jar" streamfusion-core libstreamfusion "" assert_native_payload "$core_jar" streamfusion-core libstreamfusion "" if jar tf "$core_jar" | grep -Eq '^tech/streamfusion/(kafka|parquet|format/(json|csv|raw|avro|avroconfluent|protobuf))/'; then @@ -98,7 +101,7 @@ fi for suffix in kafka json csv raw avro protobuf parquet; do assert_native_payload \ - "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-$version.jar" \ + "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-flink$flink_line-$version.jar" \ "streamfusion-$suffix" "libstreamfusion_$suffix" "$suffix" done @@ -106,10 +109,10 @@ assert_no_native_payload \ "$repo_root/streamfusion-runtime/target/streamfusion-runtime-$version.jar" \ streamfusion-runtime assert_no_native_payload \ - "$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-$version.jar" \ + "$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-flink$flink_line-$version.jar" \ streamfusion-avro-confluent-registry -loader_jar="$repo_root/streamfusion-loader/target/streamfusion-loader-$version.jar" +loader_jar="$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$version.jar" if [ ! -f "$loader_jar" ]; then echo "missing artifact: $loader_jar" >&2 exit 1 @@ -127,7 +130,7 @@ if [ -n "$duplicates" ]; then exit 1 fi -confluent_jar="$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-$version.jar" +confluent_jar="$repo_root/streamfusion-avro-confluent-registry/target/streamfusion-avro-confluent-registry-flink$flink_line-$version.jar" if jar tf "$confluent_jar" | grep -q 'libstreamfusion_avro'; then echo "the Confluent integration duplicates streamfusion-avro's native library" >&2 exit 1 diff --git a/bin/flink-suite.sh b/bin/flink-suite.sh index 84b09389..81c06032 100755 --- a/bin/flink-suite.sh +++ b/bin/flink-suite.sh @@ -5,6 +5,14 @@ set -uo pipefail readonly REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" readonly FLINK_VERSION="${FLINK_VERSION:-2.2.1}" readonly FLINK_TAG="release-${FLINK_VERSION}" +# The suite must build StreamFusion for the same Flink line it is about to run against. +readonly FLINK_LINE="${FLINK_VERSION%.*}" +readonly FLINK_KAFKA_CONNECTOR_VERSION="${KAFKA_CONNECTOR_VERSION:-5.0.0}-${FLINK_LINE}" +if [[ "${FLINK_LINE}" == "2.2" ]]; then + readonly SF_FLINK_PROFILE_ARG="" +else + readonly SF_FLINK_PROFILE_ARG="-Pflink-${FLINK_LINE}" +fi readonly KAFKA_CONNECTOR_VERSION="${KAFKA_CONNECTOR_VERSION:-5.0.0}" readonly KAFKA_CONNECTOR_TAG="v${KAFKA_CONNECTOR_VERSION}" readonly SUITE_ROOT="${FLINK_SUITE_ROOT:-${REPO_ROOT}/.flink-suite}" @@ -13,7 +21,9 @@ readonly KAFKA_CONNECTOR_ROOT="${SUITE_ROOT}/flink-connector-kafka-${KAFKA_CONNE readonly STREAMFUSION_BUILD_ROOT="${SUITE_ROOT}/streamfusion-source" readonly AGENT_ROOT="${REPO_ROOT}/dev/flink-suite/agent" readonly AGENT_JAR="${AGENT_ROOT}/target/streamfusion-flink-suite-agent-1.0-SNAPSHOT.jar" -readonly CLASSPATH_FILE="${SUITE_ROOT}/streamfusion-classpath.txt" +# Per line: the two lines resolve different Flink, Calcite and connector jars, so a shared file lets +# a reused build run one line's tests against the other line's classpath. +readonly CLASSPATH_FILE="${SUITE_ROOT}/streamfusion-classpath-${FLINK_VERSION}.txt" readonly MAVEN_SETTINGS="${REPO_ROOT}/dev/flink-suite/settings.xml" readonly SUITE_MAVEN_REPO="${SUITE_ROOT}/m2" readonly UNSHADED_PLANNER_JAR="${SUITE_ROOT}/flink-table-planner-${FLINK_VERSION}-unshaded.jar" @@ -196,13 +206,32 @@ else ) || exit $? echo "Building and installing StreamFusion and its supported connector/format modules against the source-suite planner..." + # Flink pins Calcite per line, and the source-suite classpath must agree with it: a planner + # compiled against one Calcite cannot initialise its convertlet table against another. + readonly FLINK_TABLE_POM="${FLINK_ROOT}/flink-table/pom.xml" + CALCITE_VERSION="$(sed -n 's:.*\(.*\).*:\1:p' "${FLINK_TABLE_POM}" | head -1)" + if [[ -z "${CALCITE_VERSION}" ]]; then + echo "Could not read calcite.version from ${FLINK_TABLE_POM}" >&2 + exit 1 + fi + echo "Flink ${FLINK_VERSION} pins Calcite ${CALCITE_VERSION}." + # Deployable coordinates carry the Flink line, so the module list is per-line too. + SF_MODULES="" + for module in core kafka json csv raw avro avro-confluent-registry protobuf parquet; do + SF_MODULES="${SF_MODULES}${SF_MODULES:+,}:streamfusion-${module}-flink${FLINK_LINE}" + done mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \ -Dstreamfusion.flink-source-suite \ + ${SF_FLINK_PROFILE_ARG} \ + -Dcalcite.version="${CALCITE_VERSION}" \ -f "${STREAMFUSION_BUILD_ROOT}/pom.xml" \ - -pl :streamfusion-core,:streamfusion-kafka,:streamfusion-json,:streamfusion-csv,:streamfusion-raw,:streamfusion-avro,:streamfusion-avro-confluent-registry,:streamfusion-protobuf,:streamfusion-parquet \ + -pl "${SF_MODULES}" \ -am -DskipTests clean install || exit $? mvn -B -ntp -s "${MAVEN_SETTINGS}" -Dmaven.repo.local="${SUITE_MAVEN_REPO}" \ -f "${REPO_ROOT}/dev/flink-suite/classpath-pom.xml" \ + -Dflink.version="${FLINK_VERSION}" \ + -Dflink.line="${FLINK_LINE}" \ + -Dflink.connector.kafka.version="${FLINK_KAFKA_CONNECTOR_VERSION}" \ dependency:build-classpath -Dmdep.outputFile="${CLASSPATH_FILE}" || exit $? if [[ "${SUITE_MODE}" == "formats" || "${SUITE_MODE}" == "parquet" ]]; then diff --git a/bin/install-flink.sh b/bin/install-flink.sh index c99938af..c7dcc994 100755 --- a/bin/install-flink.sh +++ b/bin/install-flink.sh @@ -2,8 +2,15 @@ set -eu +flink_line=2.2 +if [ "${1:-}" = "--flink-line" ]; then + [ "$#" -ge 2 ] || { echo "usage: $0 [--flink-line ] " >&2; exit 64; } + flink_line=$2 + shift 2 +fi + if [ "$#" -ne 1 ]; then - echo "usage: $0 " >&2 + echo "usage: $0 [--flink-line ] " >&2 exit 64 fi @@ -11,8 +18,8 @@ flink_home=$1 script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) repo_root=$(cd "$script_dir/.." && pwd) artifact_version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version) -loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-$artifact_version.jar -core_jar=$repo_root/streamfusion-core/target/streamfusion-core-$artifact_version-runtime.jar +loader_jar=$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$artifact_version.jar +core_jar=$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$artifact_version-runtime.jar if [ ! -d "$flink_home/lib" ]; then echo "Flink lib directory does not exist: $flink_home/lib" >&2 diff --git a/bin/package-release.sh b/bin/package-release.sh index 325664e1..9299ef54 100755 --- a/bin/package-release.sh +++ b/bin/package-release.sh @@ -4,10 +4,17 @@ set -eu script_dir=$(CDPATH='' cd -- "$(dirname -- "$0")" && pwd) repo_root=$(cd "$script_dir/.." && pwd) +# Deployable coordinates carry the Flink line, so a bundle holds one line's jars. +flink_line=2.2 +if [ "${1:-}" = "--flink-line" ]; then + [ "$#" -ge 2 ] || { echo "usage: $0 [--flink-line ] [output-dir]" >&2; exit 64; } + flink_line=$2 + shift 2 +fi version=$(cd "$repo_root" && mvn -q -DforceStdout help:evaluate -Dexpression=project.version) output_dir=${1:-"$repo_root/target/release"} stage_dir=$(mktemp -d) -bundle_dir=$stage_dir/streamfusion-$version +bundle_dir=$stage_dir/streamfusion-flink$flink_line-$version cleanup() { rm -rf "$stage_dir" @@ -16,15 +23,15 @@ trap cleanup EXIT HUP INT TERM mkdir -p "$bundle_dir" "$output_dir" cp "$repo_root/LICENSE" "$repo_root/readme.md" "$bundle_dir/" -cp "$repo_root/streamfusion-loader/target/streamfusion-loader-$version.jar" "$bundle_dir/" -cp "$repo_root/streamfusion-core/target/streamfusion-core-$version-runtime.jar" "$bundle_dir/" +cp "$repo_root/streamfusion-loader/target/streamfusion-loader-flink$flink_line-$version.jar" "$bundle_dir/" +cp "$repo_root/streamfusion-core/target/streamfusion-core-flink$flink_line-$version-runtime.jar" "$bundle_dir/" for suffix in kafka json csv raw avro avro-confluent-registry protobuf parquet; do - cp "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-$version.jar" "$bundle_dir/" + cp "$repo_root/streamfusion-$suffix/target/streamfusion-$suffix-flink$flink_line-$version.jar" "$bundle_dir/" done -archive=$output_dir/streamfusion-$version-bin.tar.gz -(cd "$stage_dir" && tar -czf "$archive" "streamfusion-$version") +archive=$output_dir/streamfusion-flink$flink_line-$version-bin.tar.gz +(cd "$stage_dir" && tar -czf "$archive" "streamfusion-flink$flink_line-$version") (cd "$output_dir" && shasum -a 256 "$(basename "$archive")" > "$(basename "$archive").sha256") printf '%s\n' "$archive" diff --git a/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java b/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java index 856b733f..94f7f4ab 100644 --- a/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java +++ b/dev/flink-suite/agent/src/main/java/tech/streamfusion/suite/StreamFusionSuiteAgent.java @@ -2,12 +2,18 @@ import java.lang.instrument.Instrumentation; import java.lang.reflect.InvocationTargetException; +import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; +import java.util.List; import java.util.Set; import java.util.WeakHashMap; import java.util.concurrent.atomic.AtomicBoolean; import net.bytebuddy.agent.builder.AgentBuilder; import net.bytebuddy.asm.Advice; +import net.bytebuddy.description.type.TypeDescription; +import net.bytebuddy.dynamic.DynamicType; +import net.bytebuddy.utility.JavaModule; import static net.bytebuddy.matcher.ElementMatchers.named; import static net.bytebuddy.matcher.ElementMatchers.takesArguments; @@ -32,6 +38,9 @@ public final class StreamFusionSuiteAgent { private static final AtomicBoolean NATIVE_MEMORY_STATE_REPORTED = new AtomicBoolean(); private static final AtomicBoolean ROCKSDB_STATE_REPORTED = new AtomicBoolean(); private static final AtomicBoolean NATIVE_PARQUET_WRITER_REPORTED = new AtomicBoolean(); + private static final Set TRANSFORMED_TYPES = Collections.synchronizedSet(new HashSet<>()); + private static final AtomicBoolean PLANNER_ADVICE_ENTERED = new AtomicBoolean(); + private static volatile Instrumentation INSTRUMENTATION; private static final Set INSTALLED_CONFIGS = Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>())); private static final ThreadLocal UNMODIFIED_PLAN_SETUP = new ThreadLocal<>(); @@ -41,8 +50,13 @@ public final class StreamFusionSuiteAgent { private StreamFusionSuiteAgent() {} public static void premain(String arguments, Instrumentation instrumentation) { + INSTRUMENTATION = instrumentation; + Runtime.getRuntime() + .addShutdownHook( + new Thread(StreamFusionSuiteAgent::auditInterception, "streamfusion-suite-audit")); new AgentBuilder.Default() .with(AgentBuilder.Listener.StreamWriting.toSystemError().withTransformationsOnly()) + .with(new InterceptionAudit()) .type(named(PLANNER_FACTORY)) .transform( (builder, type, classLoader, module, protectionDomain) -> @@ -124,6 +138,7 @@ private InstallStreamFusion() {} @Advice.OnMethodEnter static void enter(@Advice.Argument(0) Object context) { + StreamFusionSuiteAgent.recordPlannerAdviceEntered(); try { if (requiresUnmodifiedFlinkPlan()) { return; @@ -277,4 +292,72 @@ static void enter() { } } } + + public static void recordPlannerAdviceEntered() { + PLANNER_ADVICE_ENTERED.set(true); + } + + /** + * Fails the JVM when the planner interception never took effect. + * + *

The advice is attached by class and method name, so on an untested Flink version a rename or + * a signature change attaches nothing at all and the upstream suite then passes while running + * stock Flink — a green result that proves nothing. Loaded-but-never-instrumented is unambiguous + * and halts; instrumented-but-never-entered only warns, because a suite can load the factory + * without ever building a table environment. + */ + static void auditInterception() { + Instrumentation instrumentation = INSTRUMENTATION; + if (instrumentation == null) { + return; + } + boolean plannerLoaded = isLoaded(instrumentation, PLANNER_FACTORY); + boolean delegateLoaded = isLoaded(instrumentation, DELEGATE_PLANNER_FACTORY); + if (!plannerLoaded && !delegateLoaded) { + return; // no table stack in this JVM, so no interception was expected + } + List problems = new ArrayList<>(); + if (plannerLoaded && !TRANSFORMED_TYPES.contains(PLANNER_FACTORY)) { + problems.add(PLANNER_FACTORY + " was loaded but never instrumented"); + } + if (delegateLoaded && !TRANSFORMED_TYPES.contains(DELEGATE_PLANNER_FACTORY)) { + problems.add(DELEGATE_PLANNER_FACTORY + " was loaded but never instrumented"); + } + if (!problems.isEmpty()) { + System.err.println( + "FATAL: the StreamFusion suite agent did not instrument the Flink planner factory." + + " This run exercised stock Flink and any pass it reported is meaningless."); + problems.forEach(problem -> System.err.println(" - " + problem)); + System.err.flush(); + Runtime.getRuntime().halt(70); + } + if (!PLANNER_ADVICE_ENTERED.get()) { + System.err.println( + "WARNING: the StreamFusion suite agent instrumented the Flink planner factory, but its" + + " create(..) advice never ran — check that the method matcher still applies."); + } + } + + private static boolean isLoaded(Instrumentation instrumentation, String className) { + for (Class loaded : instrumentation.getAllLoadedClasses()) { + if (className.equals(loaded.getName())) { + return true; + } + } + return false; + } + + /** Records which interception targets actually took effect. */ + private static final class InterceptionAudit extends AgentBuilder.Listener.Adapter { + + @Override + public void onTransformation( + TypeDescription typeDescription, + ClassLoader classLoader, + JavaModule module, + boolean loaded, + DynamicType dynamicType) { + TRANSFORMED_TYPES.add(typeDescription.getName()); + } + } } diff --git a/dev/flink-suite/classpath-pom.xml b/dev/flink-suite/classpath-pom.xml index b46aefea..ae6f93b6 100644 --- a/dev/flink-suite/classpath-pom.xml +++ b/dev/flink-suite/classpath-pom.xml @@ -6,50 +6,56 @@ tech.streamfusion streamfusion-flink-suite-classpath 1.0-SNAPSHOT + + + 2.2.1 + 2.2 + 5.0.0-2.2 + tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-kafka + streamfusion-kafka-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-json + streamfusion-json-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-csv + streamfusion-csv-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-raw + streamfusion-raw-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-avro + streamfusion-avro-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-avro-confluent-registry + streamfusion-avro-confluent-registry-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-protobuf + streamfusion-protobuf-flink${flink.line} 0.1.0-rc2 tech.streamfusion - streamfusion-parquet + streamfusion-parquet-flink${flink.line} 0.1.0-rc2 2.2.1 - 4.2.0 2.12 4.32.1 + + 1.36.0 4.4.0 @@ -183,7 +192,7 @@ org.apache.flink flink-connector-kafka - 5.0.0-2.2 + ${flink.connector.kafka.version} provided + + flink-2.1 + + 2.1.3 + 2.1 + 5.0.0-2.1 + java-flink2.1 + + 3.21.7 + 1.34.0 + + @@ -617,7 +641,7 @@ org.apache.calcite calcite-core - 1.36.0 + ${calcite.version} provided diff --git a/src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java b/src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java new file mode 100644 index 00000000..06405b1a --- /dev/null +++ b/src/main/java-flink2.1/tech/streamfusion/planner/compat/FlinkCompat.java @@ -0,0 +1,183 @@ +package tech.streamfusion.planner.compat; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import javax.annotation.Nullable; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexProgram; +import org.apache.flink.api.common.functions.FlatMapFunction; +import org.apache.flink.api.dag.Transformation; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.api.functions.async.AsyncFunction; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.functions.AsyncTableFunction; +import org.apache.flink.table.functions.TableFunction; +import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator; +import org.apache.flink.table.planner.delegation.PlannerBase; +import org.apache.flink.table.planner.plan.nodes.exec.utils.TransformationMetadata; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin; +import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; +import org.apache.flink.table.planner.plan.utils.FunctionCallUtils; +import org.apache.flink.table.planner.plan.utils.FlinkRexUtil; +import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend; +import org.apache.flink.table.types.logical.RowType; + +/** + * The Flink 2.1 half of the planner seam. See the 2.2 copy for the contract; only the Flink-facing + * names and the two capabilities 2.1 does not have differ. + */ +public final class FlinkCompat { + + private FlinkCompat() {} + + // ---------------------------------------------------------------- lookup join + + public static LookupKeys lookupKeys(StreamPhysicalLookupJoin join) { + Map keys = new HashMap<>(); + scala.collection.JavaConverters.mapAsJavaMapConverter(join.allLookupKeys()) + .asJava() + .forEach((index, param) -> keys.put((Integer) index, param)); + return LookupKeys.of(keys); + } + + public static @Nullable String unsupportedKeyShape(LookupKeys keys) { + for (Object param : keys.raw().values()) { + if (!(param instanceof FunctionCallUtils.FieldRef) + && !(param instanceof FunctionCallUtils.Constant)) { + return "lookup join: unsupported lookup key shape " + param.getClass().getSimpleName(); + } + } + return null; + } + + public static @Nullable AsyncLookupOptions asyncOptions(StreamPhysicalLookupJoin join) { + if (join.asyncOptions().isEmpty()) { + return null; + } + FunctionCallUtils.AsyncOptions options = join.asyncOptions().get(); + return new AsyncLookupOptions(options.asyncBufferCapacity, options.keyOrdered); + } + + public static GeneratedAsyncFetcher generateAsyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + AsyncTableFunction lookupFunction, + String tableName) { + LookupJoinCodeGenerator.GeneratedTableFunctionWithDataType> + generated = + LookupJoinCodeGenerator.generateAsyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName); + return new GeneratedAsyncFetcher(generated.tableFunc(), generated.dataType()); + } + + public static GeneratedFunction> generateSyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + TableFunction lookupFunction, + String tableName, + boolean objectReuseEnabled) { + return LookupJoinCodeGenerator.generateSyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName, + objectReuseEnabled); + } + + public static Transformation applyCustomShufflePartitioner( + PlannerBase planner, + RelOptTable temporalTable, + RowType probeType, + LookupKeys lookupKeys, + Transformation rows, + ChangelogMode inputChangelogMode, + TransformationMetadata metadata) { + return LookupJoinUtil.tryApplyCustomShufflePartitioner( + planner, temporalTable, probeType, rawKeys(lookupKeys), rows, inputChangelogMode, metadata); + } + + private static List orderedKeys(LookupKeys lookupKeys) { + Map keys = rawKeys(lookupKeys); + List ordered = new ArrayList<>(keys.size()); + for (int key : LookupJoinUtil.getOrderedLookupKeys(keys.keySet())) { + ordered.add(keys.get(key)); + } + return ordered; + } + + @SuppressWarnings("unchecked") + private static Map rawKeys(LookupKeys lookupKeys) { + return (Map) (Map) lookupKeys.raw(); + } + + // ------------------------------------------------------- changelog normalize + + /** Flink 2.1 has no source-reuse marking pass, so a normalize never shares a source. */ + public static boolean sharesSourceOrCommonFilter(StreamPhysicalChangelogNormalize normalize) { + return false; + } + + // ------------------------------------------------------------ watermark push-down + + /** + * Flink 2.1 does not carry a rowtime expression on the pushed spec, and its watermark generator is + * generated from the watermark expression alone, so there is nothing to cross-check against. The + * caller reads the rowtime column out of the watermark expression either way. + */ + public static Optional watermarkRowtimeExpr(WatermarkPushDownSpec spec) { + return Optional.empty(); + } + + // ---------------------------------------------------------------- dimension calc + + /** Flink 2.1 returns the projection as a Scala {@code Seq}; the erased type is identical. */ + public static ExpandedCalc expandCalcProgram(RexProgram calc) { + scala.Tuple2, scala.Option> expanded = + FlinkRexUtil.expandRexProgram(calc); + return new ExpandedCalc( + scala.collection.JavaConverters.seqAsJavaListConverter(expanded._1()).asJava(), + expanded._2().isDefined() ? expanded._2().get() : null); + } + + // ----------------------------------------------------------------- state backend + + /** + * Flink 2.1's keyed-state backend has no type identifier, so nothing on that line consumes this; + * the value names the backend this one delegates to. + */ + public static String backendTypeIdentifier(CheckpointableKeyedStateBackend delegate) { + return "rocksdb"; + } +} diff --git a/src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java b/src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java new file mode 100644 index 00000000..f713f7df --- /dev/null +++ b/src/main/java-flink2.2/tech/streamfusion/planner/compat/FlinkCompat.java @@ -0,0 +1,189 @@ +package tech.streamfusion.planner.compat; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import javax.annotation.Nullable; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexProgram; +import org.apache.flink.api.common.functions.FlatMapFunction; +import org.apache.flink.api.dag.Transformation; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.streaming.api.functions.async.AsyncFunction; +import org.apache.flink.table.catalog.DataTypeFactory; +import org.apache.flink.table.connector.ChangelogMode; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.functions.AsyncTableFunction; +import org.apache.flink.table.functions.TableFunction; +import org.apache.flink.table.planner.codegen.FunctionCallCodeGenerator; +import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator; +import org.apache.flink.table.planner.delegation.PlannerBase; +import org.apache.flink.table.planner.plan.nodes.exec.utils.TransformationMetadata; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin; +import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; +import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; +import org.apache.flink.table.planner.plan.utils.FlinkRexUtil; +import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend; +import org.apache.flink.table.types.logical.RowType; + +/** + * The Flink 2.2 half of the planner seam. + * + *

Everything Flink changed between the supported minor versions is reached through this class, so + * the rest of the planner compiles once against a single source tree. Only the file is swapped per + * Flink line — the signatures below are the contract both copies satisfy. + */ +public final class FlinkCompat { + + private FlinkCompat() {} + + // ---------------------------------------------------------------- lookup join + + /** Flink 2.2 renamed {@code FunctionCallUtils} to {@code FunctionCallUtil}; members are equal. */ + public static LookupKeys lookupKeys(StreamPhysicalLookupJoin join) { + Map keys = new HashMap<>(); + scala.collection.JavaConverters.mapAsJavaMapConverter(join.allLookupKeys()) + .asJava() + .forEach((index, param) -> keys.put((Integer) index, param)); + return LookupKeys.of(keys); + } + + /** Returns a decline reason when any key is not a plain field reference or constant. */ + public static @Nullable String unsupportedKeyShape(LookupKeys keys) { + for (Object param : keys.raw().values()) { + if (!(param instanceof FunctionCallUtil.FieldRef) + && !(param instanceof FunctionCallUtil.Constant)) { + return "lookup join: unsupported lookup key shape " + param.getClass().getSimpleName(); + } + } + return null; + } + + public static @Nullable AsyncLookupOptions asyncOptions(StreamPhysicalLookupJoin join) { + if (join.asyncOptions().isEmpty()) { + return null; + } + FunctionCallUtil.AsyncOptions options = join.asyncOptions().get(); + return new AsyncLookupOptions(options.asyncBufferCapacity, options.keyOrdered); + } + + public static GeneratedAsyncFetcher generateAsyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + AsyncTableFunction lookupFunction, + String tableName) { + FunctionCallCodeGenerator.GeneratedTableFunctionWithDataType> + generated = + LookupJoinCodeGenerator.generateAsyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName); + return new GeneratedAsyncFetcher(generated.tableFunc(), generated.dataType()); + } + + public static GeneratedFunction> generateSyncFetcher( + ReadableConfig config, + ClassLoader classLoader, + DataTypeFactory dataTypeFactory, + RowType probeType, + RowType tableSourceRowType, + RowType resultRowType, + LookupKeys lookupKeys, + TableFunction lookupFunction, + String tableName, + boolean objectReuseEnabled) { + return LookupJoinCodeGenerator.generateSyncLookupFunction( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + orderedKeys(lookupKeys), + lookupFunction, + tableName, + objectReuseEnabled); + } + + /** The connector-owned partitioning SPI, which is typed on the renamed parameter map. */ + public static Transformation applyCustomShufflePartitioner( + PlannerBase planner, + RelOptTable temporalTable, + RowType probeType, + LookupKeys lookupKeys, + Transformation rows, + ChangelogMode inputChangelogMode, + TransformationMetadata metadata) { + return LookupJoinUtil.tryApplyCustomShufflePartitioner( + planner, temporalTable, probeType, rawKeys(lookupKeys), rows, inputChangelogMode, metadata); + } + + private static List orderedKeys(LookupKeys lookupKeys) { + Map keys = rawKeys(lookupKeys); + List ordered = new ArrayList<>(keys.size()); + for (int key : LookupJoinUtil.getOrderedLookupKeys(keys.keySet())) { + ordered.add(keys.get(key)); + } + return ordered; + } + + @SuppressWarnings("unchecked") + private static Map rawKeys(LookupKeys lookupKeys) { + return (Map) (Map) lookupKeys.raw(); + } + + // ------------------------------------------------------- changelog normalize + + /** + * Whether the rel carries the source-reuse marking Flink 2.2 added. The 2.2 optimizer runs a + * {@code FlinkMarkChangelogNormalizeProgram} pass that can share one normalize across reused + * sources and hoist a common filter; the native operator reproduces neither. + */ + public static boolean sharesSourceOrCommonFilter(StreamPhysicalChangelogNormalize normalize) { + return normalize.sourceReused() || normalize.commonFilter().length > 0; + } + + // ------------------------------------------------------------ watermark push-down + + /** + * The rowtime expression Flink 2.2 carries alongside the watermark expression. Flink 2.1 keeps + * only the watermark expression, so the caller falls back to deriving the rowtime field itself. + */ + public static Optional watermarkRowtimeExpr(WatermarkPushDownSpec spec) { + return spec.getRowtimeExpr(); + } + + // ---------------------------------------------------------------- dimension calc + + /** Flink 2.2 returns the projection as a {@code java.util.List}. */ + public static ExpandedCalc expandCalcProgram(RexProgram calc) { + scala.Tuple2, scala.Option> expanded = + FlinkRexUtil.expandRexProgram(calc); + return new ExpandedCalc( + expanded._1(), expanded._2().isDefined() ? expanded._2().get() : null); + } + + // ----------------------------------------------------------------- state backend + + /** Reported through the keyed-state backend interface from Flink 2.2 on. */ + public static String backendTypeIdentifier(CheckpointableKeyedStateBackend delegate) { + return delegate.getBackendTypeIdentifier(); + } +} diff --git a/src/main/java/tech/streamfusion/arrow/ArrowConversion.java b/src/main/java/tech/streamfusion/arrow/ArrowConversion.java index 65abcf17..44b84ba3 100644 --- a/src/main/java/tech/streamfusion/arrow/ArrowConversion.java +++ b/src/main/java/tech/streamfusion/arrow/ArrowConversion.java @@ -18,48 +18,12 @@ package tech.streamfusion.arrow; -import tech.streamfusion.arrow.vectors.ArrowArrayColumnVector; -import tech.streamfusion.arrow.vectors.ArrowBigIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowBinaryColumnVector; -import tech.streamfusion.arrow.vectors.ArrowBooleanColumnVector; -import tech.streamfusion.arrow.vectors.ArrowDateColumnVector; -import tech.streamfusion.arrow.vectors.ArrowDecimalColumnVector; -import tech.streamfusion.arrow.vectors.ArrowDoubleColumnVector; -import tech.streamfusion.arrow.vectors.ArrowFloatColumnVector; -import tech.streamfusion.arrow.vectors.ArrowIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowMapColumnVector; -import tech.streamfusion.arrow.vectors.ArrowNullColumnVector; -import tech.streamfusion.arrow.vectors.ArrowRowColumnVector; -import tech.streamfusion.arrow.vectors.ArrowSmallIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowTimeColumnVector; -import tech.streamfusion.arrow.vectors.ArrowTimestampColumnVector; -import tech.streamfusion.arrow.vectors.ArrowTinyIntColumnVector; -import tech.streamfusion.arrow.vectors.ArrowVarBinaryColumnVector; -import tech.streamfusion.arrow.vectors.ArrowVarCharColumnVector; -import tech.streamfusion.arrow.writers.ArrayWriter; -import tech.streamfusion.arrow.writers.ArrowFieldWriter; -import tech.streamfusion.arrow.writers.BigIntWriter; -import tech.streamfusion.arrow.writers.BinaryWriter; -import tech.streamfusion.arrow.writers.BooleanWriter; -import tech.streamfusion.arrow.writers.DateWriter; -import tech.streamfusion.arrow.writers.DecimalWriter; -import tech.streamfusion.arrow.writers.DoubleWriter; -import tech.streamfusion.arrow.writers.FloatWriter; -import tech.streamfusion.arrow.writers.IntWriter; -import tech.streamfusion.arrow.writers.MapWriter; -import tech.streamfusion.arrow.writers.NullWriter; -import tech.streamfusion.arrow.writers.RowWriter; -import tech.streamfusion.arrow.writers.SmallIntWriter; -import tech.streamfusion.arrow.writers.TimeWriter; -import tech.streamfusion.arrow.writers.TimestampWriter; -import tech.streamfusion.arrow.writers.TinyIntWriter; -import tech.streamfusion.arrow.writers.VarBinaryWriter; -import tech.streamfusion.arrow.writers.VarCharWriter; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.stream.Collectors; + import org.apache.arrow.vector.BigIntVector; import org.apache.arrow.vector.BitVector; import org.apache.arrow.vector.DateDayVector; @@ -69,6 +33,7 @@ import org.apache.arrow.vector.Float4Vector; import org.apache.arrow.vector.Float8Vector; import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.IntervalDayVector; import org.apache.arrow.vector.NullVector; import org.apache.arrow.vector.SmallIntVector; import org.apache.arrow.vector.TimeMicroVector; @@ -100,8 +65,8 @@ import org.apache.flink.table.types.logical.BooleanType; import org.apache.flink.table.types.logical.CharType; import org.apache.flink.table.types.logical.DateType; -import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DayTimeIntervalType; +import org.apache.flink.table.types.logical.DecimalType; import org.apache.flink.table.types.logical.DoubleType; import org.apache.flink.table.types.logical.FloatType; import org.apache.flink.table.types.logical.IntType; @@ -119,6 +84,45 @@ import org.apache.flink.table.types.logical.YearMonthIntervalType; import org.apache.flink.table.types.logical.utils.LogicalTypeDefaultVisitor; +import tech.streamfusion.arrow.vectors.ArrowArrayColumnVector; +import tech.streamfusion.arrow.vectors.ArrowBigIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowBinaryColumnVector; +import tech.streamfusion.arrow.vectors.ArrowBooleanColumnVector; +import tech.streamfusion.arrow.vectors.ArrowDateColumnVector; +import tech.streamfusion.arrow.vectors.ArrowDecimalColumnVector; +import tech.streamfusion.arrow.vectors.ArrowDoubleColumnVector; +import tech.streamfusion.arrow.vectors.ArrowFloatColumnVector; +import tech.streamfusion.arrow.vectors.ArrowIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowIntervalDayColumnVector; +import tech.streamfusion.arrow.vectors.ArrowMapColumnVector; +import tech.streamfusion.arrow.vectors.ArrowNullColumnVector; +import tech.streamfusion.arrow.vectors.ArrowRowColumnVector; +import tech.streamfusion.arrow.vectors.ArrowSmallIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowTimeColumnVector; +import tech.streamfusion.arrow.vectors.ArrowTimestampColumnVector; +import tech.streamfusion.arrow.vectors.ArrowTinyIntColumnVector; +import tech.streamfusion.arrow.vectors.ArrowVarBinaryColumnVector; +import tech.streamfusion.arrow.vectors.ArrowVarCharColumnVector; +import tech.streamfusion.arrow.writers.ArrayWriter; +import tech.streamfusion.arrow.writers.ArrowFieldWriter; +import tech.streamfusion.arrow.writers.BigIntWriter; +import tech.streamfusion.arrow.writers.BinaryWriter; +import tech.streamfusion.arrow.writers.BooleanWriter; +import tech.streamfusion.arrow.writers.DateWriter; +import tech.streamfusion.arrow.writers.DecimalWriter; +import tech.streamfusion.arrow.writers.DoubleWriter; +import tech.streamfusion.arrow.writers.FloatWriter; +import tech.streamfusion.arrow.writers.IntWriter; +import tech.streamfusion.arrow.writers.MapWriter; +import tech.streamfusion.arrow.writers.NullWriter; +import tech.streamfusion.arrow.writers.RowWriter; +import tech.streamfusion.arrow.writers.SmallIntWriter; +import tech.streamfusion.arrow.writers.TimeWriter; +import tech.streamfusion.arrow.writers.TimestampWriter; +import tech.streamfusion.arrow.writers.TinyIntWriter; +import tech.streamfusion.arrow.writers.VarBinaryWriter; +import tech.streamfusion.arrow.writers.VarCharWriter; + /** * The Arrow ↔ {@link RowData} type mapping, reader factory, and writer factory, ported (and trimmed) from * Flink's {@code org.apache.flink.table.runtime.arrow.ArrowUtils}. Vendored rather than depended on @@ -278,6 +282,8 @@ static ColumnVector createColumnVector(ValueVector vector, LogicalType fieldType || vector instanceof TimeMicroVector || vector instanceof TimeNanoVector) { return new ArrowTimeColumnVector(vector); + } else if (vector instanceof IntervalDayVector) { + return new ArrowIntervalDayColumnVector((IntervalDayVector) vector); } else if (vector instanceof TimeStampVector) { return new ArrowTimestampColumnVector(vector); } else if (vector instanceof MapVector) { @@ -306,7 +312,11 @@ static ColumnVector createColumnVector(ValueVector vector, LogicalType fieldType } else if (vector instanceof NullVector) { return ArrowNullColumnVector.INSTANCE; } else { - throw new UnsupportedOperationException(String.format("Unsupported type %s.", fieldType)); + throw new UnsupportedOperationException(String.format( + "Unsupported type %s (Arrow vector %s, arrow type %s).", + fieldType, + vector.getClass().getSimpleName(), + vector.getField().getType())); } } @@ -374,7 +384,11 @@ private static ArrowFieldWriter createArrowFieldWriterForRow( } else if (vector instanceof NullVector) { return new NullWriter<>((NullVector) vector); } else { - throw new UnsupportedOperationException(String.format("Unsupported type %s.", fieldType)); + throw new UnsupportedOperationException(String.format( + "Unsupported type %s (Arrow vector %s, arrow type %s).", + fieldType, + vector.getClass().getSimpleName(), + vector.getField().getType())); } } @@ -444,7 +458,11 @@ private static ArrowFieldWriter createArrowFieldWriterForArray( } else if (vector instanceof NullVector) { return new NullWriter<>((NullVector) vector); } else { - throw new UnsupportedOperationException(String.format("Unsupported type %s.", fieldType)); + throw new UnsupportedOperationException(String.format( + "Unsupported type %s (Arrow vector %s, arrow type %s).", + fieldType, + vector.getClass().getSimpleName(), + vector.getField().getType())); } } diff --git a/src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java b/src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java new file mode 100644 index 00000000..d0209ae2 --- /dev/null +++ b/src/main/java/tech/streamfusion/arrow/vectors/ArrowIntervalDayColumnVector.java @@ -0,0 +1,56 @@ +/* + * 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 tech.streamfusion.arrow.vectors; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.data.columnar.vector.LongColumnVector; +import org.apache.flink.util.Preconditions; + +import org.apache.arrow.vector.IntervalDayVector; + +/** + * Arrow column vector for a day-time INTERVAL carried as Arrow's own {@code Interval(DAY_TIME)}. + * + *

StreamFusion's canonical form for these is a signed millisecond {@code Int64} — Flink's internal + * representation — but DataFusion produces a native interval array when an expression's result type + * is an interval rather than a timestamp. Both encodings therefore reach this boundary, exactly as + * the four Arrow time encodings do for {@code TIME}. + */ +@Internal +public final class ArrowIntervalDayColumnVector implements LongColumnVector { + + private static final long MILLIS_PER_DAY = 86_400_000L; + + private final IntervalDayVector valueVector; + + public ArrowIntervalDayColumnVector(IntervalDayVector valueVector) { + this.valueVector = Preconditions.checkNotNull(valueVector); + } + + @Override + public long getLong(int i) { + return IntervalDayVector.getDays(valueVector.getDataBuffer(), i) * MILLIS_PER_DAY + + IntervalDayVector.getMilliseconds(valueVector.getDataBuffer(), i); + } + + @Override + public boolean isNullAt(int i) { + return valueVector.isNull(i); + } +} diff --git a/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java b/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java index 59e014ce..69024b17 100644 --- a/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java +++ b/src/main/java/tech/streamfusion/planner/ChangelogNormalizeMatcher.java @@ -1,6 +1,7 @@ package tech.streamfusion.planner; import tech.streamfusion.operator.RowDataArrowConverter; +import tech.streamfusion.planner.compat.FlinkCompat; import org.apache.calcite.rel.RelNode; import org.apache.flink.table.planner.calcite.FlinkTypeFactory$; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalChangelogNormalize; @@ -23,7 +24,7 @@ static boolean matches(StreamPhysicalChangelogNormalize node) { if (node.filterCondition() != null) { return false; // a pushed filter condition is not yet reproduced } - if (node.sourceReused() || node.commonFilter().length > 0) { + if (FlinkCompat.sharesSourceOrCommonFilter(node)) { return false; // the source-reuse rewrite changes the operator's contract } return RowDataArrowConverter.supports( @@ -42,7 +43,7 @@ static String unsupportedReason(StreamPhysicalChangelogNormalize node) { if (node.filterCondition() != null) { return "changelog normalize: a pushed filter condition is not supported"; } - if (node.sourceReused() || node.commonFilter().length > 0) { + if (FlinkCompat.sharesSourceOrCommonFilter(node)) { return "changelog normalize: the source-reuse variant is not supported"; } return "changelog normalize: needs a row type the Arrow conversion supports"; diff --git a/src/main/java/tech/streamfusion/planner/HostCastFunction.java b/src/main/java/tech/streamfusion/planner/HostCastFunction.java index bf3eb571..e1c7464e 100644 --- a/src/main/java/tech/streamfusion/planner/HostCastFunction.java +++ b/src/main/java/tech/streamfusion/planner/HostCastFunction.java @@ -88,12 +88,49 @@ private void initializeExecutor(ClassLoader classLoader) { // java.lang.reflect switches from its native accessor to a generated accessor after a small // invocation threshold. Force that transition while Flink's job classloader is open; otherwise // a long-running native batch can cross the threshold after the safety wrapper was retired. + Object sample = warmupValue(inputType); for (int i = 0; i < 20; i++) { - executor.cast(null); + try { + executor.cast(sample); + } catch (Throwable warmupFailure) { + // Warming is an optimization, never a precondition: a rejected sample must not fail startup. + break; + } } } } + /** + * A value the executor can actually consume. {@code null} is not legal input for a NOT NULL type, + * whose generated cast dereferences the argument without a guard. + */ + private static Object warmupValue(LogicalType type) { + switch (type.getTypeRoot()) { + case CHAR: + case VARCHAR: + return StringData.fromString("0"); + case BOOLEAN: + return Boolean.FALSE; + case TINYINT: + return (byte) 0; + case SMALLINT: + return (short) 0; + case INTEGER: + return 0; + case BIGINT: + return 0L; + case FLOAT: + return 0f; + case DOUBLE: + return 0d; + case DECIMAL: + DecimalType decimal = (DecimalType) type; + return DecimalData.fromBigDecimal(BigDecimal.ZERO, decimal.getPrecision(), decimal.getScale()); + default: + return null; + } + } + /** The upcall marshals external values (String/BigDecimal/boxed numbers); the executor speaks * Flink's internal data. */ private Object toInternal(Object value) { diff --git a/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java b/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java index 313efb12..d3f58b26 100644 --- a/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java +++ b/src/main/java/tech/streamfusion/planner/LookupJoinMatcher.java @@ -1,13 +1,12 @@ package tech.streamfusion.planner; -import java.util.HashMap; -import java.util.Map; import org.apache.calcite.plan.RelOptTable; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.core.JoinRelType; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLookupJoin; import org.apache.flink.table.planner.plan.schema.TableSourceTable; -import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; +import tech.streamfusion.planner.compat.FlinkCompat; +import tech.streamfusion.planner.compat.LookupKeys; /** * Recognizes the processing-time lookup joins the native operator runs: {@code probe JOIN dim FOR @@ -40,22 +39,12 @@ static String unsupportedReason(StreamPhysicalLookupJoin join) { if (!(unwrapTable(join.temporalTable()) instanceof TableSourceTable)) { return "lookup join: temporal table is not a (non-legacy) table source"; } - for (FunctionCallUtil.FunctionParam param : lookupKeys(join).values()) { - if (!(param instanceof FunctionCallUtil.FieldRef) - && !(param instanceof FunctionCallUtil.Constant)) { - return "lookup join: unsupported lookup key shape " + param.getClass().getSimpleName(); - } - } - return null; + return FlinkCompat.unsupportedKeyShape(lookupKeys(join)); } /** The dimension key → probe field/constant map the generated fetcher builds its key row from. */ - static Map lookupKeys(StreamPhysicalLookupJoin join) { - Map keys = new HashMap<>(); - scala.collection.JavaConverters.mapAsJavaMapConverter(join.allLookupKeys()) - .asJava() - .forEach((index, param) -> keys.put((Integer) index, param)); - return keys; + static LookupKeys lookupKeys(StreamPhysicalLookupJoin join) { + return FlinkCompat.lookupKeys(join); } static boolean isLeftOuterJoin(StreamPhysicalLookupJoin join) { @@ -85,7 +74,7 @@ static RelNode substitute(StreamPhysicalLookupJoin join, PlanContext ctx) { join.finalPreFilterCondition().isDefined() ? join.finalPreFilterCondition().get() : null, join.finalRemainingCondition().isDefined() ? join.finalRemainingCondition().get() : null, LookupJoinMatcher.isLeftOuterJoin(join), - join.asyncOptions().isDefined() ? join.asyncOptions().get() : null, + FlinkCompat.asyncOptions(join), join.retryOptions().isDefined() ? join.retryOptions().get() : null, join.preferCustomShuffle(), join.inputChangelogMode()); diff --git a/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java b/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java index e594cb89..4ecdee0d 100644 --- a/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java +++ b/src/main/java/tech/streamfusion/planner/NativeLookupJoinExecNode.java @@ -6,10 +6,12 @@ import tech.streamfusion.operator.NativeAsyncLookupJoinOperator; import tech.streamfusion.operator.NativeLookupJoinOperator; import tech.streamfusion.operator.RowDataToArrowOperator; -import java.util.ArrayList; +import tech.streamfusion.planner.compat.AsyncLookupOptions; +import tech.streamfusion.planner.compat.FlinkCompat; +import tech.streamfusion.planner.compat.GeneratedAsyncFetcher; +import tech.streamfusion.planner.compat.LookupKeys; import java.util.Collections; import java.util.List; -import java.util.Map; import java.util.Optional; import javax.annotation.Nullable; import org.apache.calcite.plan.RelOptTable; @@ -19,7 +21,6 @@ import org.apache.flink.api.dag.Transformation; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.configuration.ReadableConfig; -import org.apache.flink.streaming.api.functions.async.AsyncFunction; import org.apache.flink.streaming.api.operators.OneInputStreamOperator; import org.apache.flink.table.catalog.DataTypeFactory; import org.apache.flink.table.connector.ChangelogMode; @@ -32,7 +33,6 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory; import org.apache.flink.table.planner.codegen.CodeGeneratorContext; import org.apache.flink.table.planner.codegen.FilterCodeGenerator; -import org.apache.flink.table.planner.codegen.FunctionCallCodeGenerator; import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator; import org.apache.flink.table.planner.delegation.PlannerBase; import org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase; @@ -42,7 +42,6 @@ import org.apache.flink.table.planner.plan.nodes.exec.SingleTransformationTranslator; import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode; import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil; -import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; import org.apache.flink.table.planner.utils.JavaScalaConversionUtil; import org.apache.flink.table.planner.utils.ShortcutUtils; @@ -81,13 +80,13 @@ public class NativeLookupJoinExecNode extends ExecNodeBase private final RelOptTable temporalTable; private final RowType probeType; - private final Map lookupKeys; + private final LookupKeys lookupKeys; private final @Nullable List projectionOnTemporalTable; private final @Nullable RexNode filterOnTemporalTable; private final @Nullable RexNode preFilterCondition; private final @Nullable RexNode remainingJoinCondition; private final boolean leftOuterJoin; - private final @Nullable FunctionCallUtil.AsyncOptions asyncOptions; + private final @Nullable AsyncLookupOptions asyncOptions; private final @Nullable LookupJoinUtil.RetryLookupOptions retryOptions; private final boolean preferCustomShuffle; private final ChangelogMode inputChangelogMode; @@ -99,13 +98,13 @@ public NativeLookupJoinExecNode( String description, RelOptTable temporalTable, RowType probeType, - Map lookupKeys, + LookupKeys lookupKeys, @Nullable List projectionOnTemporalTable, @Nullable RexNode filterOnTemporalTable, @Nullable RexNode preFilterCondition, @Nullable RexNode remainingJoinCondition, boolean leftOuterJoin, - @Nullable FunctionCallUtil.AsyncOptions asyncOptions, + @Nullable AsyncLookupOptions asyncOptions, @Nullable LookupJoinUtil.RetryLookupOptions retryOptions, boolean preferCustomShuffle, ChangelogMode inputChangelogMode) { @@ -144,17 +143,13 @@ protected Transformation translateToPlanInternal( RowType resultRowType = (RowType) getOutputType(); String tableName = String.join(".", temporalTable.getQualifiedName()); - List orderedKeys = new ArrayList<>(lookupKeys.size()); - for (int key : LookupJoinUtil.getOrderedLookupKeys(lookupKeys.keySet())) { - orderedKeys.add(lookupKeys.get(key)); - } boolean async = asyncOptions != null; ResultRetryStrategy retryStrategy = retryOptions == null ? ResultRetryStrategy.NO_RETRY_STRATEGY : retryOptions.toRetryStrategy(); UserDefinedFunction lookupFunction = LookupJoinUtil.getLookupFunction( temporalTable, - lookupKeys.keySet(), + lookupKeys.indexes(), classLoader, async, retryStrategy, @@ -173,7 +168,7 @@ protected Transformation translateToPlanInternal( input.getParallelism(), false); rows = - LookupJoinUtil.tryApplyCustomShufflePartitioner( + FlinkCompat.applyCustomShufflePartitioner( planner, temporalTable, probeType, @@ -222,18 +217,17 @@ protected Transformation translateToPlanInternal( OneInputStreamOperator operator; if (async) { - FunctionCallCodeGenerator.GeneratedTableFunctionWithDataType> - generatedFetcher = - LookupJoinCodeGenerator.generateAsyncLookupFunction( - config, - classLoader, - dataTypeFactory, - probeType, - tableSourceRowType, - resultRowType, - orderedKeys, - (AsyncTableFunction) lookupFunction, - tableName); + GeneratedAsyncFetcher generatedFetcher = + FlinkCompat.generateAsyncFetcher( + config, + classLoader, + dataTypeFactory, + probeType, + tableSourceRowType, + resultRowType, + lookupKeys, + (AsyncTableFunction) lookupFunction, + tableName); GeneratedResultFuture> generatedResultFuture = LookupJoinCodeGenerator.generateTableAsyncCollector( config, @@ -249,35 +243,35 @@ protected Transformation translateToPlanInternal( AsyncLookupJoinRunner runner = generatedCalc != null ? new AsyncLookupJoinWithCalcRunner( - generatedFetcher.tableFunc(), + generatedFetcher.tableFunction(), fetcherConverter, generatedCalc, generatedResultFuture, generatedPreFilter, InternalSerializers.create(rightRowType), leftOuterJoin, - asyncOptions.asyncBufferCapacity) + asyncOptions.bufferCapacity()) : new AsyncLookupJoinRunner( - generatedFetcher.tableFunc(), + generatedFetcher.tableFunction(), fetcherConverter, generatedResultFuture, generatedPreFilter, InternalSerializers.create(rightRowType), leftOuterJoin, - asyncOptions.asyncBufferCapacity); + asyncOptions.bufferCapacity()); operator = new NativeAsyncLookupJoinOperator( - runner, probeType, resultRowType, asyncOptions.keyOrdered); + runner, probeType, resultRowType, asyncOptions.keyOrdered()); } else { GeneratedFunction> generatedFetcher = - LookupJoinCodeGenerator.generateSyncLookupFunction( + FlinkCompat.generateSyncFetcher( config, classLoader, dataTypeFactory, probeType, tableSourceRowType, resultRowType, - orderedKeys, + lookupKeys, (TableFunction) lookupFunction, tableName, planner.getExecEnv().getConfig().isObjectReuseEnabled()); diff --git a/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java b/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java index bdf8f525..a4659834 100644 --- a/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java +++ b/src/main/java/tech/streamfusion/planner/PhysicalPlanScan.java @@ -22,6 +22,7 @@ import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalGroupAggregate; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalGroupWindowAggregate; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalIntervalJoin; +import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeltaJoin; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalJoin; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLimit; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalLocalGroupAggregate; @@ -91,6 +92,14 @@ private RelNode optimizeConfigured(RelNode root) { LOG.info("StreamFusion native acceleration is disabled; the plan runs on Flink"); return root; } + // Flink runs its FORCE delta-join validation after this pass, and only complains when a regular + // join survives. Substituting one away would silently run a plan the host means to reject. + if (deltaJoinForceWouldReject(root)) { + fallbackReasons.add( + "delta join: table.optimizer.delta-join.strategy is FORCE but the plan has no delta join"); + LOG.info("StreamFusion declined the plan so Flink can enforce its FORCE delta-join strategy"); + return root; + } RelNode optimized = substitute(root); // The one always-on plan-time summary; -Dstreamfusion.logFallbackReasons=true itemizes the // reasons and explainSummary() carries them into explain output. @@ -721,6 +730,29 @@ private void record(RelNode node) { } } + /** Mirrors {@code StreamPhysicalDeltaJoinForceValidator}, which spares a plan that has any delta join. */ + private static boolean deltaJoinForceWouldReject(RelNode root) { + if (ShortcutUtils.unwrapTableConfig(root) + .get(OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY) + != OptimizerConfigOptions.DeltaJoinStrategy.FORCE) { + return false; + } + return contains(root, StreamPhysicalJoin.class) + && !contains(root, StreamPhysicalDeltaJoin.class); + } + + private static boolean contains(RelNode node, Class relType) { + if (relType.isInstance(node)) { + return true; + } + for (RelNode input : node.getInputs()) { + if (contains(input, relType)) { + return true; + } + } + return false; + } + /** Operator types seen in the optimized physical plans, in traversal order. */ public List operatorTypes() { return List.copyOf(operatorTypes); diff --git a/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java b/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java index 0d0bfb38..0810bba8 100644 --- a/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java +++ b/src/main/java/tech/streamfusion/planner/ScanWatermarkSpec.java @@ -12,6 +12,7 @@ import org.apache.flink.table.planner.plan.abilities.source.SourceAbilitySpec; import org.apache.flink.table.planner.plan.abilities.source.SourceWatermarkSpec; import org.apache.flink.table.planner.plan.abilities.source.WatermarkPushDownSpec; +import tech.streamfusion.planner.compat.FlinkCompat; import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalTableSourceScan; import org.apache.flink.table.planner.plan.schema.TableSourceTable; import org.apache.flink.table.planner.utils.ShortcutUtils; @@ -75,8 +76,9 @@ static ScanWatermarkSpec of(StreamPhysicalTableSourceScan scan) { // computed rowtime); it must be one of the supported terms and agree with the watermark // expression's column. Integer rowtimeFromExpr = null; - if (pushed.getRowtimeExpr().isPresent()) { - Integer index = rowtimeTerm(stripReinterpret(pushed.getRowtimeExpr().get())); + var declaredRowtime = FlinkCompat.watermarkRowtimeExpr(pushed); + if (declaredRowtime.isPresent()) { + Integer index = rowtimeTerm(stripReinterpret(declaredRowtime.get())); if (index == null) { return UNSUPPORTED; } diff --git a/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java b/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java index 128f6938..fd85300b 100644 --- a/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java +++ b/src/main/java/tech/streamfusion/planner/StreamPhysicalNativeLookupJoin.java @@ -1,7 +1,6 @@ package tech.streamfusion.planner; import java.util.List; -import java.util.Map; import javax.annotation.Nullable; import org.apache.calcite.plan.RelOptCluster; import org.apache.calcite.plan.RelOptTable; @@ -13,11 +12,13 @@ import org.apache.flink.table.planner.calcite.FlinkTypeFactory$; import org.apache.flink.table.planner.plan.nodes.exec.ExecNode; import org.apache.flink.table.planner.plan.nodes.exec.InputProperty; -import org.apache.flink.table.planner.plan.utils.FlinkRexUtil; -import org.apache.flink.table.planner.plan.utils.FunctionCallUtil; import org.apache.flink.table.planner.plan.utils.LookupJoinUtil; import org.apache.flink.table.planner.utils.ShortcutUtils; import org.apache.flink.table.connector.ChangelogMode; +import tech.streamfusion.planner.compat.AsyncLookupOptions; +import tech.streamfusion.planner.compat.ExpandedCalc; +import tech.streamfusion.planner.compat.FlinkCompat; +import tech.streamfusion.planner.compat.LookupKeys; /** * Physical node standing in for a processing-time lookup join the native operator runs. Columnar on @@ -31,12 +32,12 @@ public class StreamPhysicalNativeLookupJoin extends StreamPhysicalNativeSingleRe implements ColumnarInput, ColumnarOutput { private final RelOptTable temporalTable; - private final Map lookupKeys; + private final LookupKeys lookupKeys; private final @Nullable RexProgram calcOnTemporalTable; private final @Nullable RexNode preFilterCondition; private final @Nullable RexNode remainingJoinCondition; private final boolean leftOuterJoin; - private final @Nullable FunctionCallUtil.AsyncOptions asyncOptions; + private final @Nullable AsyncLookupOptions asyncOptions; private final @Nullable LookupJoinUtil.RetryLookupOptions retryOptions; private final boolean preferCustomShuffle; private final ChangelogMode inputChangelogMode; @@ -47,12 +48,12 @@ public StreamPhysicalNativeLookupJoin( RelNode input, RelDataType outputRowType, RelOptTable temporalTable, - Map lookupKeys, + LookupKeys lookupKeys, @Nullable RexProgram calcOnTemporalTable, @Nullable RexNode preFilterCondition, @Nullable RexNode remainingJoinCondition, boolean leftOuterJoin, - @Nullable FunctionCallUtil.AsyncOptions asyncOptions, + @Nullable AsyncLookupOptions asyncOptions, @Nullable LookupJoinUtil.RetryLookupOptions retryOptions, boolean preferCustomShuffle, ChangelogMode inputChangelogMode) { @@ -100,10 +101,9 @@ public ExecNode translateToExecNode() { List projectionOnTemporalTable = null; RexNode filterOnTemporalTable = null; if (calcOnTemporalTable != null) { - scala.Tuple2, scala.Option> expanded = - FlinkRexUtil.expandRexProgram(calcOnTemporalTable); - projectionOnTemporalTable = expanded._1(); - filterOnTemporalTable = expanded._2().isDefined() ? expanded._2().get() : null; + ExpandedCalc expanded = FlinkCompat.expandCalcProgram(calcOnTemporalTable); + projectionOnTemporalTable = expanded.projection(); + filterOnTemporalTable = expanded.filter(); } return new NativeLookupJoinExecNode( ShortcutUtils.unwrapTableConfig(this), diff --git a/src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java b/src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java new file mode 100644 index 00000000..ebb324be --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/AsyncLookupOptions.java @@ -0,0 +1,27 @@ +package tech.streamfusion.planner.compat; + +/** + * The async-lookup settings a lookup join was planned with, carried opaquely. + * + *

Flink renamed the enclosing utility between 2.1 and 2.2 but kept the fields, so the two values + * the native operator needs are copied out here and the original is retained only for the codegen + * call that still requires it. + */ +public final class AsyncLookupOptions { + + private final int bufferCapacity; + private final boolean keyOrdered; + + public AsyncLookupOptions(int bufferCapacity, boolean keyOrdered) { + this.bufferCapacity = bufferCapacity; + this.keyOrdered = keyOrdered; + } + + public int bufferCapacity() { + return bufferCapacity; + } + + public boolean keyOrdered() { + return keyOrdered; + } +} diff --git a/src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java b/src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java new file mode 100644 index 00000000..f74e1d44 --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/ExpandedCalc.java @@ -0,0 +1,31 @@ +package tech.streamfusion.planner.compat; + +import java.util.List; +import javax.annotation.Nullable; +import org.apache.calcite.rex.RexNode; + +/** + * A dimension-side calc split into its projection and optional filter. + * + *

Flink returns the projection as a Scala {@code Seq} on 2.1 and a {@code java.util.List} on 2.2. + * The two erase to the same descriptor, so the difference is invisible to bytecode comparison and + * only shows up when compiling — hence this holder rather than the raw tuple. + */ +public final class ExpandedCalc { + + private final List projection; + private final @Nullable RexNode filter; + + public ExpandedCalc(List projection, @Nullable RexNode filter) { + this.projection = projection; + this.filter = filter; + } + + public List projection() { + return projection; + } + + public @Nullable RexNode filter() { + return filter; + } +} diff --git a/src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java b/src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java new file mode 100644 index 00000000..83e11b91 --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/GeneratedAsyncFetcher.java @@ -0,0 +1,32 @@ +package tech.streamfusion.planner.compat; + +import org.apache.flink.streaming.api.functions.async.AsyncFunction; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.generated.GeneratedFunction; +import org.apache.flink.table.types.DataType; + +/** + * The generated async lookup fetcher and the data type its results arrive in. + * + *

Flink moved the class that pairs these two between 2.1 and 2.2; both members are unchanged, so + * they are carried here instead and shared planner code never names the moved type. + */ +public final class GeneratedAsyncFetcher { + + private final GeneratedFunction> tableFunction; + private final DataType dataType; + + public GeneratedAsyncFetcher( + GeneratedFunction> tableFunction, DataType dataType) { + this.tableFunction = tableFunction; + this.dataType = dataType; + } + + public GeneratedFunction> tableFunction() { + return tableFunction; + } + + public DataType dataType() { + return dataType; + } +} diff --git a/src/main/java/tech/streamfusion/planner/compat/LookupKeys.java b/src/main/java/tech/streamfusion/planner/compat/LookupKeys.java new file mode 100644 index 00000000..cefc3aff --- /dev/null +++ b/src/main/java/tech/streamfusion/planner/compat/LookupKeys.java @@ -0,0 +1,42 @@ +package tech.streamfusion.planner.compat; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; + +/** + * The dimension-key → probe field/constant map a lookup join builds its key row from, carried + * opaquely. + * + *

Flink renamed the enclosing utility (and therefore the parameter type) between 2.1 and 2.2 + * without changing any member. Shared planner code passes this holder around and never names the + * Flink type; {@code FlinkCompat} — the one class compiled per Flink line — is the only place that + * unwraps it. + */ +public final class LookupKeys { + + private final Map byIndex; + + private LookupKeys(Map byIndex) { + this.byIndex = Collections.unmodifiableMap(byIndex); + } + + /** Wraps Flink's already-extracted parameter map. Called only from {@code FlinkCompat}. */ + public static LookupKeys of(Map byIndex) { + return new LookupKeys(new LinkedHashMap<>(byIndex)); + } + + public Set indexes() { + return byIndex.keySet(); + } + + public int size() { + return byIndex.size(); + } + + /** The raw Flink parameters. Callers outside {@code FlinkCompat} must treat these as opaque. */ + public Map raw() { + return byIndex; + } +} diff --git a/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java b/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java index 0f4c8c37..766b35ea 100644 --- a/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java +++ b/src/main/java/tech/streamfusion/state/RocksDBNativeKeyedStateBackend.java @@ -432,8 +432,9 @@ public boolean isSafeToReuseKVState() { return delegateUnchecked().isSafeToReuseKVState(); } - @Override + // Declared on the backend interface only from Flink 2.2; unused on 2.1. public String getBackendTypeIdentifier() { - return delegateUnchecked().getBackendTypeIdentifier(); + return tech.streamfusion.planner.compat.FlinkCompat.backendTypeIdentifier( + delegateUnchecked()); } } diff --git a/src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java b/src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java new file mode 100644 index 00000000..46d503df --- /dev/null +++ b/src/test/java/tech/streamfusion/EnabledIfFlinkAtLeast.java @@ -0,0 +1,26 @@ +package tech.streamfusion; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; +import org.junit.jupiter.api.extension.ExtendWith; + +/** + * Skips a test whose behaviour the Flink under test cannot exhibit yet. + * + *

Reserved for cases where the host itself lacks the behaviour, so there is no Flink result to + * be identical to. A StreamFusion coverage gap must never be hidden behind this. + */ +@Target({ElementType.TYPE, ElementType.METHOD}) +@Retention(RetentionPolicy.RUNTIME) +@ExtendWith(FlinkVersionCondition.class) +@interface EnabledIfFlinkAtLeast { + + int major(); + + int minor(); + + /** Why the older line cannot run it, ideally the upstream issue key. */ + String reason(); +} diff --git a/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java b/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java index 699eac20..94411a83 100644 --- a/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java +++ b/src/test/java/tech/streamfusion/FlinkUnnestSqlHarnessTest.java @@ -78,6 +78,10 @@ void leftUnnestMatchesHost() throws Exception { } @Test + @EnabledIfFlinkAtLeast( + major = 2, + minor = 2, + reason = "FLINK-33217; earlier planners cannot type this query at all") void leftUnnestWithOrdinalityMatchesHost() throws Exception { // A LEFT null-pad row carries a null ordinal too. NativeParity.assertParity( diff --git a/src/test/java/tech/streamfusion/FlinkVersionCondition.java b/src/test/java/tech/streamfusion/FlinkVersionCondition.java new file mode 100644 index 00000000..29910939 --- /dev/null +++ b/src/test/java/tech/streamfusion/FlinkVersionCondition.java @@ -0,0 +1,30 @@ +package tech.streamfusion; + +import java.util.Optional; +import org.junit.jupiter.api.extension.ConditionEvaluationResult; +import org.junit.jupiter.api.extension.ExecutionCondition; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.junit.platform.commons.support.AnnotationSupport; + +class FlinkVersionCondition implements ExecutionCondition { + + @Override + public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) { + Optional required = + AnnotationSupport.findAnnotation(context.getElement(), EnabledIfFlinkAtLeast.class); + if (required.isEmpty()) { + return ConditionEvaluationResult.enabled("no Flink version requirement"); + } + EnabledIfFlinkAtLeast annotation = required.get(); + if (HostFlinkVersion.atLeast(annotation.major(), annotation.minor())) { + return ConditionEvaluationResult.enabled("host Flink is new enough"); + } + return ConditionEvaluationResult.disabled( + "needs Flink %d.%d or newer (%s); host is %s" + .formatted( + annotation.major(), + annotation.minor(), + annotation.reason(), + HostFlinkVersion.current())); + } +} diff --git a/src/test/java/tech/streamfusion/HostFlinkVersion.java b/src/test/java/tech/streamfusion/HostFlinkVersion.java new file mode 100644 index 00000000..3fd5e67c --- /dev/null +++ b/src/test/java/tech/streamfusion/HostFlinkVersion.java @@ -0,0 +1,25 @@ +package tech.streamfusion; + +import org.apache.flink.runtime.util.EnvironmentInformation; + +/** + * The Flink line the tests are executing against. + * + *

Read from the running Flink rather than a build property so a test can never be gated on a + * version different from the one it actually loaded. + */ +final class HostFlinkVersion { + + private HostFlinkVersion() {} + + static String current() { + return EnvironmentInformation.getVersion(); + } + + static boolean atLeast(int major, int minor) { + String[] parts = current().split("[.-]"); + int hostMajor = Integer.parseInt(parts[0]); + int hostMinor = Integer.parseInt(parts[1]); + return hostMajor != major ? hostMajor > major : hostMinor >= minor; + } +} diff --git a/src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java b/src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java new file mode 100644 index 00000000..d79af622 --- /dev/null +++ b/src/test/java/tech/streamfusion/arrow/ArrowIntervalDayColumnVectorTest.java @@ -0,0 +1,61 @@ +package tech.streamfusion.arrow; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Collections; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.IntervalDayVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.IntervalUnit; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.FieldType; +import org.apache.arrow.vector.types.pojo.Schema; +import org.apache.flink.table.types.logical.DayTimeIntervalType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.junit.jupiter.api.Test; + +/** + * A day-time INTERVAL reaches the boundary either as the Int64 millis StreamFusion canonicalises on, + * or as Arrow's own {@code Interval(DAY_TIME)} when the native result type is an interval rather than + * a timestamp. Both must read back as Flink's internal millisecond long. + */ +class ArrowIntervalDayColumnVectorTest { + + private static final RowType SCHEMA = + RowType.of( + new LogicalType[] { + new DayTimeIntervalType(DayTimeIntervalType.DayTimeResolution.SECOND) + }, + new String[] {"i"}); + + @Test + void readsArrowDayTimeIntervalAsMillis() { + Field field = + new Field( + "i", + FieldType.nullable(new ArrowType.Interval(IntervalUnit.DAY_TIME)), + Collections.emptyList()); + try (BufferAllocator allocator = new RootAllocator(); + VectorSchemaRoot root = + VectorSchemaRoot.create(new Schema(Collections.singletonList(field)), allocator)) { + IntervalDayVector vector = (IntervalDayVector) root.getVector("i"); + vector.allocateNew(3); + vector.set(0, 0, 6_000); + // The days component must be folded in, not dropped. + vector.set(1, 2, 500); + vector.setNull(2); + vector.setValueCount(3); + root.setRowCount(3); + + ArrowReader reader = ArrowConversion.createArrowReader(root, SCHEMA); + + assertEquals(6_000L, reader.read(0).getLong(0)); + assertEquals(2 * 86_400_000L + 500L, reader.read(1).getLong(0)); + assertTrue(reader.read(2).isNullAt(0)); + } + } +} diff --git a/src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java b/src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java new file mode 100644 index 00000000..5088bb72 --- /dev/null +++ b/src/test/java/tech/streamfusion/planner/DeltaJoinForceGateTest.java @@ -0,0 +1,115 @@ +package tech.streamfusion.planner; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; +import org.apache.flink.table.api.config.OptimizerConfigOptions; +import org.apache.flink.types.Row; +import org.junit.jupiter.api.Test; + +/** + * Flink validates its FORCE delta-join strategy after our pass runs, and rejects a plan only when a + * regular join survived. Substituting that join away would turn a query the host means to reject + * into one that silently runs, so the pass declines such plans wholesale. The gate mirrors the + * host's condition exactly rather than declining on FORCE alone — acceleration is kept for every + * plan the validator would have passed. + */ +class DeltaJoinForceGateTest { + + private static final String JOIN_QUERY = + "SELECT a.k, a.v, b.w FROM A AS a JOIN B AS b ON a.k = b.k"; + + @Test + void forceWithoutADeltaJoinLeavesThePlanForFlinkToReject() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + tEnv.getConfig() + .set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + OptimizerConfigOptions.DeltaJoinStrategy.FORCE); + + ValidationException failure = + assertThrows(ValidationException.class, () -> tEnv.explainSql(JOIN_QUERY)); + + assertTrue( + failure.getMessage().contains("delta join"), + "expected Flink's own FORCE rejection, got: " + failure.getMessage()); + assertEquals(0, scan.substitutions(), scan::explainSummary); + assertTrue( + scan.fallbackReasons().stream().anyMatch(reason -> reason.startsWith("delta join:")), + "the decline must be reported as a fallback reason, saw: " + scan.fallbackReasons()); + } + + @Test + void forceStillAcceleratesAPlanWithoutAJoin() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + tEnv.getConfig() + .set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + OptimizerConfigOptions.DeltaJoinStrategy.FORCE); + + tEnv.explainSql("SELECT k, v * 2 FROM A"); + + assertTrue(scan.substitutions() > 0, scan::explainSummary); + } + + @Test + void theDefaultStrategyAcceleratesTheSameJoin() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + + tEnv.explainSql(JOIN_QUERY); + + assertTrue(scan.substitutions() > 0, scan::explainSummary); + } + + @Test + void noneStrategyAcceleratesTheSameJoin() { + TableEnvironment tEnv = environment(); + PhysicalPlanScan scan = NativePlanner.install(tEnv); + tEnv.getConfig() + .set( + OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY, + OptimizerConfigOptions.DeltaJoinStrategy.NONE); + + tEnv.explainSql(JOIN_QUERY); + + assertTrue(scan.substitutions() > 0, scan::explainSummary); + } + + private static TableEnvironment environment() { + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + env.setParallelism(1); + StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); + + DataStream a = + env.fromData( + Types.ROW_NAMED(new String[] {"k", "v"}, Types.LONG, Types.LONG), + Row.of(1L, 10L), + Row.of(2L, 20L)); + DataStream b = + env.fromData( + Types.ROW_NAMED(new String[] {"k", "w"}, Types.LONG, Types.LONG), + Row.of(1L, 100L), + Row.of(2L, 200L)); + tEnv.createTemporaryView( + "A", + a, + Schema.newBuilder().column("k", DataTypes.BIGINT()).column("v", DataTypes.BIGINT()).build()); + tEnv.createTemporaryView( + "B", + b, + Schema.newBuilder().column("k", DataTypes.BIGINT()).column("w", DataTypes.BIGINT()).build()); + return tEnv; + } +} diff --git a/src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java b/src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java new file mode 100644 index 00000000..7e5398f7 --- /dev/null +++ b/src/test/java/tech/streamfusion/planner/HostCastFunctionTest.java @@ -0,0 +1,57 @@ +package tech.streamfusion.planner; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.List; +import org.apache.flink.table.types.logical.BigIntType; +import org.apache.flink.table.types.logical.CharType; +import org.apache.flink.table.types.logical.DoubleType; +import org.apache.flink.table.types.logical.FloatType; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.SmallIntType; +import org.apache.flink.table.types.logical.TinyIntType; +import org.apache.flink.table.types.logical.VarCharType; +import org.junit.jupiter.api.Test; + +class HostCastFunctionTest { + + private static final List NOT_NULL_STRINGS = + List.of( + new CharType(false, 3), + new VarCharType(false, 5), + new VarCharType(false, VarCharType.MAX_LENGTH)); + + private static final List NUMBERS = + List.of( + new TinyIntType(), + new SmallIntType(), + new IntType(), + new BigIntType(), + new FloatType(), + new DoubleType()); + + /** + * A NOT NULL input type's generated cast has no null guard — it trims the argument directly — so + * warming the executor with a null failed the operator's open() instead of any row. + */ + @Test + void opensForNotNullStringToNumberCasts() { + for (LogicalType input : NOT_NULL_STRINGS) { + for (LogicalType target : NUMBERS) { + assertDoesNotThrow( + () -> new HostCastFunction(input, target).open(null), input + " -> " + target); + } + } + } + + /** Warming must not consume the executor: the first real row still casts. */ + @Test + void castsAfterWarmup() { + HostCastFunction function = + new HostCastFunction(new VarCharType(false, 5), new IntType()); + function.open(null); + assertEquals(-7, function.eval("-7")); + } +} diff --git a/streamfusion-avro-confluent-registry/pom.xml b/streamfusion-avro-confluent-registry/pom.xml index 372674ad..3a23b951 100644 --- a/streamfusion-avro-confluent-registry/pom.xml +++ b/streamfusion-avro-confluent-registry/pom.xml @@ -2,12 +2,12 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-avro-confluent-registry + streamfusion-avro-confluent-registry-flink${flink.line} StreamFusion Avro Confluent Registry true - tech.streamfusionstreamfusion-core${project.version}provided - tech.streamfusionstreamfusion-avro${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided + tech.streamfusionstreamfusion-avro-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resourcesorg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/avroconfluent/**/*.javatech/streamfusion/kafka/ConfluentSchemaRegistry.java diff --git a/streamfusion-avro/pom.xml b/streamfusion-avro/pom.xml index 4bf24a92..8d729049 100644 --- a/streamfusion-avro/pom.xml +++ b/streamfusion-avro/pom.xml @@ -2,9 +2,9 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-avro + streamfusion-avro-flink${flink.line} StreamFusion Avro true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resources${project.basedir}/../native/target/universal/avrotech/streamfusion/native/avro**/libstreamfusion_avro.so**/libstreamfusion_avro.dyliborg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/avro/**/*.java diff --git a/streamfusion-core/pom.xml b/streamfusion-core/pom.xml index d97c304b..9e729264 100644 --- a/streamfusion-core/pom.xml +++ b/streamfusion-core/pom.xml @@ -10,12 +10,31 @@ ${revision} - streamfusion-core + streamfusion-core-flink${flink.line} StreamFusion Core ${project.basedir}/../src/main/java + + + org.codehaus.mojo + build-helper-maven-plugin + 3.6.0 + + + add-flink-compat-source + generate-sources + add-source + + + ${project.basedir}/../src/main/${flink.compat.source} + + + + + org.apache.maven.plugins maven-compiler-plugin diff --git a/streamfusion-csv/pom.xml b/streamfusion-csv/pom.xml index f0eab1b0..ffc0d8d3 100644 --- a/streamfusion-csv/pom.xml +++ b/streamfusion-csv/pom.xml @@ -2,9 +2,9 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-csv + streamfusion-csv-flink${flink.line} StreamFusion CSV true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resources${project.basedir}/../native/target/universal/csvtech/streamfusion/native/csv**/libstreamfusion_csv.so**/libstreamfusion_csv.dyliborg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/csv/**/*.java diff --git a/streamfusion-delta/pom.xml b/streamfusion-delta/pom.xml index 650b3d95..05530f44 100644 --- a/streamfusion-delta/pom.xml +++ b/streamfusion-delta/pom.xml @@ -8,7 +8,7 @@ StreamFusion ${revision} - streamfusion-delta + streamfusion-delta-flink${flink.line} StreamFusion Delta true @@ -21,35 +21,35 @@ tech.streamfusion - streamfusion-kafka + streamfusion-kafka-flink${flink.line} ${project.version} test tech.streamfusion - streamfusion-json + streamfusion-json-flink${flink.line} ${project.version} test tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} provided tech.streamfusion - streamfusion-parquet + streamfusion-parquet-flink${flink.line} ${project.version} provided io.delta - delta-flink_2.2 + delta-flink_${flink.line} ${delta.version} provided - org.apache.logging.log4j diff --git a/streamfusion-image-it/pom.xml b/streamfusion-image-it/pom.xml index 5d305b92..0b7f2f63 100644 --- a/streamfusion-image-it/pom.xml +++ b/streamfusion-image-it/pom.xml @@ -48,6 +48,7 @@ ${project.build.directory}/${project.build.finalName}.jar ${project.basedir}/.. ${project.version} + ${flink.line} diff --git a/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/NativeExtensionJarIT.java b/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/NativeExtensionJarIT.java index 874b2971..133e2c0d 100644 --- a/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/NativeExtensionJarIT.java +++ b/streamfusion-image-it/src/test/java/tech/streamfusion/imageit/NativeExtensionJarIT.java @@ -50,7 +50,8 @@ private static Process extensionProcess(String extension) throws IOException { ExtensionProbe.class.getName(), requiredProperty("streamfusion.project.dir"), requiredProperty("streamfusion.version"), - extension) + extension, + requiredProperty("streamfusion.flink.line")) .redirectErrorStream(true); process.environment().put("GLIBC_TUNABLES", "glibc.rtld.optional_static_tls=131072"); return process.start(); @@ -65,9 +66,11 @@ public static void main(String[] args) throws Exception { Path projectDirectory = Path.of(args[0]); String version = args[1]; String extension = args[2]; - Path core = artifact(projectDirectory, "streamfusion-core", version); - Path extensionJar = artifact(projectDirectory, "streamfusion-" + extension, version); - URL[] classpath = extensionClasspath(projectDirectory, version, extension, core, extensionJar); + String flinkLine = args[3]; + Path core = artifact(projectDirectory, "streamfusion-core", version, flinkLine); + Path extensionJar = artifact(projectDirectory, "streamfusion-" + extension, version, flinkLine); + URL[] classpath = + extensionClasspath(projectDirectory, version, extension, core, extensionJar, flinkLine); try (URLClassLoader loader = new URLClassLoader( classpath, ClassLoader.getPlatformClassLoader())) { @@ -84,10 +87,15 @@ public static void main(String[] args) throws Exception { } private static URL[] extensionClasspath( - Path projectDirectory, String version, String extension, Path core, Path extensionJar) + Path projectDirectory, + String version, + String extension, + Path core, + Path extensionJar, + String flinkLine) throws IOException { if ("avro-confluent-registry".equals(extension)) { - Path avro = artifact(projectDirectory, "streamfusion-avro", version); + Path avro = artifact(projectDirectory, "streamfusion-avro", version, flinkLine); return new URL[] {core.toUri().toURL(), avro.toUri().toURL(), extensionJar.toUri().toURL()}; } if ("parquet".equals(extension)) { @@ -101,12 +109,13 @@ private static URL[] extensionClasspath( return new URL[] {core.toUri().toURL(), extensionJar.toUri().toURL()}; } - private static Path artifact(Path projectDirectory, String module, String version) { + private static Path artifact( + Path projectDirectory, String module, String version, String flinkLine) { Path artifact = projectDirectory .resolve(module) .resolve("target") - .resolve(module + "-" + version + ".jar"); + .resolve(module + "-flink" + flinkLine + "-" + version + ".jar"); if (!Files.isRegularFile(artifact)) { throw new IllegalStateException("Missing packaged extension artifact: " + artifact); } diff --git a/streamfusion-json/pom.xml b/streamfusion-json/pom.xml index 10acbcb0..caa35565 100644 --- a/streamfusion-json/pom.xml +++ b/streamfusion-json/pom.xml @@ -4,10 +4,10 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-json + streamfusion-json-flink${flink.line} StreamFusion JSON true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java diff --git a/streamfusion-kafka/pom.xml b/streamfusion-kafka/pom.xml index e978c330..cbb2c943 100644 --- a/streamfusion-kafka/pom.xml +++ b/streamfusion-kafka/pom.xml @@ -10,7 +10,7 @@ ${revision} - streamfusion-kafka + streamfusion-kafka-flink${flink.line} StreamFusion Kafka true @@ -18,7 +18,7 @@ tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} provided diff --git a/streamfusion-loader/pom.xml b/streamfusion-loader/pom.xml index 6317b8d3..5172ad43 100644 --- a/streamfusion-loader/pom.xml +++ b/streamfusion-loader/pom.xml @@ -5,7 +5,7 @@ 4.0.0 tech.streamfusion - streamfusion-loader + streamfusion-loader-flink${flink.line} ${revision} StreamFusion Loader @@ -48,6 +48,9 @@ admitted planner ABI patch versions. --> 2.2.0 2.2.1 + 2.2 + 5.0.0-2.2 + java-flink2.2 5.10.2 @@ -62,7 +65,7 @@ are separate lib-directory JARs, matching Flink's connector packaging model. --> tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} runtime @@ -99,7 +102,7 @@ org.apache.flink flink-connector-kafka - 5.0.0-2.2 + ${flink.connector.kafka.version} test @@ -130,6 +133,25 @@ + + org.codehaus.mojo + build-helper-maven-plugin + 3.6.0 + + + add-flink-compat-source + generate-sources + + add-source + + + + ${project.basedir}/src/main/${flink.compat.source} + + + + + org.apache.maven.plugins maven-dependency-plugin @@ -145,7 +167,7 @@ tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} runtime streamfusion-planner.jar @@ -171,6 +193,18 @@ + + + flink-2.1 + + 2.1.3 + 2.1.3 + 2.1 + 5.0.0-2.1 + java-flink2.1 + + diff --git a/streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java b/streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java new file mode 100644 index 00000000..e208a3d4 --- /dev/null +++ b/streamfusion-loader/src/main/java-flink2.1/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java @@ -0,0 +1,14 @@ +package org.apache.flink.table.planner.loader; + +import java.util.Set; + +/** + * The Flink patch versions this loader build has been validated against. See the 2.2 copy for the + * contract. + */ +final class SupportedFlinkVersions { + + static final Set VERSIONS = Set.of("2.1.3"); + + private SupportedFlinkVersions() {} +} diff --git a/streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java b/streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java new file mode 100644 index 00000000..52a52cb3 --- /dev/null +++ b/streamfusion-loader/src/main/java-flink2.2/org/apache/flink/table/planner/loader/SupportedFlinkVersions.java @@ -0,0 +1,17 @@ +package org.apache.flink.table.planner.loader; + +import java.util.Set; + +/** + * The Flink patch versions this loader build has been validated against. + * + *

The loader shadows a Flink-internal class name, so it fails closed rather than cross an + * unverified planner ABI. The set is per Flink line and must only list versions the parity and + * upstream suites have actually run against. + */ +final class SupportedFlinkVersions { + + static final Set VERSIONS = Set.of("2.2.0", "2.2.1"); + + private SupportedFlinkVersions() {} +} diff --git a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java index 9733c07f..0325fe48 100644 --- a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java +++ b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModule.java @@ -55,10 +55,10 @@ * other planner behavior continues through Flink's normal implementation. */ @Internal -public class PlannerModule { +public class PlannerModule extends PlannerModuleCompat { static final String FLINK_TABLE_PLANNER_FAT_JAR = "flink-table-planner.jar"; - private static final Set SUPPORTED_FLINK_VERSIONS = Set.of("2.2.0", "2.2.1"); + private static final Set SUPPORTED_FLINK_VERSIONS = SupportedFlinkVersions.VERSIONS; private static final String STREAMFUSION_PLANNER_JAR = "streamfusion-planner.jar"; private static final String[] STREAMFUSION_EXTENSION_PREFIXES = { "streamfusion-kafka-", @@ -129,6 +129,7 @@ private PlannerModule() { } } + @Override public URLClassLoader getSubmoduleClassLoader() { return submoduleClassLoader; } diff --git a/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java new file mode 100644 index 00000000..80c320ae --- /dev/null +++ b/streamfusion-loader/src/main/java/org/apache/flink/table/planner/loader/PlannerModuleCompat.java @@ -0,0 +1,35 @@ +/* + * 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.flink.table.planner.loader; + +import org.apache.flink.annotation.Internal; + +/** + * Carries the Flink 2.1 shape of {@code getSubmoduleClassLoader}. + * + *

Flink 2.2 narrowed the return type to {@code URLClassLoader}, so its call sites are compiled + * against a descriptor Flink 2.1's are not: 2.1 calls {@code ()Ljava/lang/ClassLoader;} and 2.2 + * calls {@code ()Ljava/net/URLClassLoader;}. Declaring the wider type here and narrowing it in the + * subclass makes the compiler emit a bridge, so the loader exposes both descriptors and satisfies + * either line from one artifact. + */ +@Internal +abstract class PlannerModuleCompat { + + public abstract ClassLoader getSubmoduleClassLoader(); +} diff --git a/streamfusion-parquet/pom.xml b/streamfusion-parquet/pom.xml index b6c9c513..e11d0634 100644 --- a/streamfusion-parquet/pom.xml +++ b/streamfusion-parquet/pom.xml @@ -10,14 +10,14 @@ ${revision} - streamfusion-parquet + streamfusion-parquet-flink${flink.line} StreamFusion Parquet true tech.streamfusion - streamfusion-core + streamfusion-core-flink${flink.line} ${project.version} provided diff --git a/streamfusion-protobuf/pom.xml b/streamfusion-protobuf/pom.xml index 7198f567..1b311ef7 100644 --- a/streamfusion-protobuf/pom.xml +++ b/streamfusion-protobuf/pom.xml @@ -2,9 +2,9 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-protobuf + streamfusion-protobuf-flink${flink.line} StreamFusion Protobuf true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resources${project.basedir}/../native/target/universal/protobuftech/streamfusion/native/protobuf**/libstreamfusion_protobuf.so**/libstreamfusion_protobuf.dyliborg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/protobuf/**/*.javatech/streamfusion/planner/ProtobufDescriptors.java diff --git a/streamfusion-raw/pom.xml b/streamfusion-raw/pom.xml index 2fd2ce01..7e2737d8 100644 --- a/streamfusion-raw/pom.xml +++ b/streamfusion-raw/pom.xml @@ -2,9 +2,9 @@ 4.0.0 tech.streamfusionStreamFusion${revision} - streamfusion-raw + streamfusion-raw-flink${flink.line} StreamFusion Raw true - tech.streamfusionstreamfusion-core${project.version}provided + tech.streamfusionstreamfusion-core-flink${flink.line}${project.version}provided ${project.basedir}/../src/main/java${project.basedir}/src/main/resources${project.basedir}/../native/target/universal/rawtech/streamfusion/native/raw**/libstreamfusion_raw.so**/libstreamfusion_raw.dyliborg.apache.maven.pluginsmaven-compiler-plugin3.13.0tech/streamfusion/format/raw/**/*.java diff --git a/streamfusion-runtime/pom.xml b/streamfusion-runtime/pom.xml index 911a7139..ca448b51 100644 --- a/streamfusion-runtime/pom.xml +++ b/streamfusion-runtime/pom.xml @@ -13,6 +13,11 @@ streamfusion-runtime StreamFusion Runtime + + true + ${project.basedir}/../src/main/java @@ -42,6 +47,25 @@ + + + org.codehaus.mojo + build-helper-maven-plugin + 3.6.0 + + + add-flink-compat-source + generate-sources + add-source + + + ${project.basedir}/../src/main/${flink.compat.source} + + + + + org.apache.maven.plugins maven-compiler-plugin