This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 541ed1c26637 feat(trino): track trinodb/trino master via a pinned SHA
(#19642)
541ed1c26637 is described below
commit 541ed1c266372928ca40d13d6f011cc1f4e639aa
Author: voonhous <[email protected]>
AuthorDate: Mon Aug 17 21:02:18 2026 +0800
feat(trino): track trinodb/trino master via a pinned SHA (#19642)
* feat(trino): track trinodb/trino master via a pinned SHA (#19640)
hudi-trino stops pinning released Trino 481 and compiles against
trinodb/trino
master at the commit in the new trino.sha root pom property (484-SNAPSHOT at
5b82ec9e7116). Trino publishes no SNAPSHOT artifacts, so scripts/trino/
bootstrap_trino.sh builds the needed modules from source into the local m2;
CI restores them from an actions cache keyed on the sha, and the nightly
compat
job now pre-seeds that cache and proposes pin advances via a bot/trino-pin
PR
(human-merged) instead of only alarming on drift (#19379).
SPI fixups 481 -> pin: split SPI rewrite (ConnectorSplitBatch removed,
getNextBatch(int, DynamicFilterSnapshot), getSplits with Set<ColumnHandle>,
wait timeout via getRequestedDynamicFilterWaitTimeoutMillis),
createPageSource
gains Optional<ConnectorTableCredentials>, TypeSignature -> TypeDescriptor,
CacheKeyProvider returns CacheKey, jts-core is now SPI surface (provided in
module and shim), Hive3MinioDataLake -> Hive3FlociDataLake/Floci in tests,
dead HudiSplit.getSplitInfo removed.
E2E keeps the released server image (new trino.e2e.version property, also
feeding trino-jdbc) and auto-skips when the pin has SPI-relevant drift from
it;
the shim plugin dir is now unzipped from the packaging zip
(trino-maven-plugin
24 emits only the zip). Release branches pin back to a released Trino; the
release guide gains the pin-back steps.
* test(trino): adapt cache tests to the blob-cache SPI at the pin (#19640)
The pin's blob-cache migration moved cache managers out of catalog
properties: fs.cache.directories/max-sizes no longer exist as catalog
config, caches come from host-loaded managers, and the cache span is
BlobCache.get instead of FileSystemCache.cacheInput.
- HudiQueryRunner loads the memory blob cache manager unconditionally,
mirroring IcebergQueryRunner; the alluxio tests register the alluxio
manager via withBlobCache + AlluxioBlobCachePlugin (TestHudiSmokeTest
grows a getBlobCacheProperties hook for the caching smoke subclass).
- TestHudiMemoryCacheFileOperations expects BlobCache.get and its span
filter includes BlobCache.* spans.
- TestTupleDomainUtilsTest wraps varchar values in Slices.utf8Slice:
TypeUtils.writeNativeValue no longer coerces raw Strings (482).
- New test deps trino-blob-cache-memory/-alluxio (BOM-managed); added
to the bootstrap module list; trino-m2 cache key bumped to v2 since
the cached artifact set changed.
All five affected test classes pass locally (95 tests) on JDK 25
against the pinned 484-SNAPSHOT artifacts.
* review(19642): address round-1 review
- hudi_trino_compat.yml: split into a read-only build job (runs mvnw from
the
trino checkout with contents: read) and a propose-pin-advance job that
holds
the only push credential and runs no third-party code; drop the bot
gh pr create (GITHUB_TOKEN PRs start no workflow runs, so required checks
would never report) in favor of pushing bot/trino-pin plus a step-summary
one-liner for a committer; title type becomes chore(trino); the pin
advance
now also refreshes trino.e2e.version from the latest released tag.
- hudi_trino_e2e.yml: drift gate switched from the compare API (300-file
cap,
saturated by a single release cycle) to uncapped per-path commit queries;
scripts/trino/** added to both paths lists.
- release_guide.md: pin-back moved to Cut a release branch, before the
source
tarball is generated (the voted RC must not ship a -SNAPSHOT trino pin);
added an SPI-surface scope re-check (jts-core/482) to the pin-back steps.
- hudi-trino/README.md: fast-iteration loop fixed (version read + unzip
moved
into step 3, which the loop repeats); pin ownership/cadence documented.
- TestHudiSmokeTest: new testDynamicFilterEliminatesAllSplits covering the
isNone() -> finished early-termination in the rewritten HudiSplitSource.
- Root pom: pin comment names the nightly job as the e2e-version owner.
* review(19642): address round-2 review
- HudiSplitSource.isFinished now reports unfinished while trinoException is
pending: the failure callback finishes the queue, and claiming finished at
that point let the engine stop polling and end the scan silently instead
of
surfacing the split-loading failure on the next getNextBatch.
- testDynamicFilterEliminatesAllSplits now pins probe-side 'Input: 0 rows'
via
EXPLAIN ANALYZE; the bare empty-result assert also passed when the filter
never engaged and the join simply discarded every row.
- e2e drift gate: lib/trino-filesystem dropped (ships inside the plugin dir,
cannot skew the boot); the gate keys on the two boundary-crossing
surfaces,
core/trino-spi and the HdfsFileSystemLoader contract
(trino-filesystem-manager
+ the server image's hdfs jar set). API responses are assigned to a
variable
before iterating so a transient gh error aborts the step instead of
reading
error JSON as drift.
- build_image.sh derives its default server version from the root pom's
trino.e2e.version (the literal default would rot when the bot bumps the
pom).
- hudi_trino_compat.yml gains a trino_ref dispatch input so the release
pin-back can verify and pin the released tag exactly; the guide's step 3
claim now matches what the workflow does.
* review(19642): address round-3 review
- HudiBackgroundSplitLoader's whenAllComplete combiner surfaces generator
failures before finishing the queue: Guava does not order the per-future
exception callbacks against the combiner, so a failure in the last future
could finish the queue before trinoException was set and isFinished would
answer true with no pending exception.
- New TestHudiSplitSource unit tests (via a package-private constructor seam
that keeps the production error-listener wiring): NONE snapshot terminates
the source with undrained splits, a loader failure blocks isFinished and
surfaces on the next batch, and normal completion still finishes.
- e2e drift gate: lib/trino-filesystem restored and lib/trino-memory-context
added -- HdfsClassLoader delegates exactly those packages from the
server's
hdfs jars back to the plugin's bundled copies, so bundling is why they
cross
the boundary; comment rewritten accordingly. Read-pin steps in ci and e2e
fail loudly on an empty sed result (an empty sha would query the default
branch and skip the suite green).
- All three trino-m2 cache keys gain the bootstrap-script hash so a
module-list
change invalidates caches without a pin move.
- compat job now also runs the hudi-trino test suite against trino HEAD
(test-source and behavioral drift blocked pin advances invisibly before);
drift-issue handler covers test failures; JDK 17 leg installs the two
extra
client modules the tests need.
- release guide no longer asks for a build_image.sh literal edit (the script
reads trino.e2e.version from the root pom since the round-2 fix).
* review(19642): sync guide pin-back paths with the e2e drift gate
---
.github/workflows/hudi_trino_ci.yml | 47 ++++--
.github/workflows/hudi_trino_compat.yml | 125 +++++++++++++--
.github/workflows/hudi_trino_e2e.yml | 137 +++++++++++++++--
docker/README.md | 11 +-
...r-compose_hadoop340_hive2310_spark402_amd64.yml | 2 +-
...r-compose_hadoop340_hive2310_spark402_arm64.yml | 2 +-
docker/trino/Dockerfile | 2 +-
docker/trino/build_image.sh | 26 ++--
docker/trino/etc/jvm.config | 2 +-
docker/trino/shim/pom.xml | 18 ++-
hudi-trino/README.md | 40 +++--
hudi-trino/pom.xml | 18 +++
.../trino/plugin/hudi/HudiPageSourceProvider.java | 3 +
.../main/java/io/trino/plugin/hudi/HudiSplit.java | 11 --
.../io/trino/plugin/hudi/HudiSplitManager.java | 6 +-
.../java/io/trino/plugin/hudi/HudiSplitSource.java | 155 ++++++++++---------
.../plugin/hudi/cache/HudiCacheKeyProvider.java | 8 +-
.../hudi/split/HudiBackgroundSplitLoader.java | 14 ++
.../trino/plugin/hudi/util/HudiAvroSerializer.java | 2 +-
.../java/io/trino/plugin/hudi/HudiQueryRunner.java | 37 ++---
.../hudi/TestHudiAlluxioCacheFileOperations.java | 8 +-
.../hudi/TestHudiAlluxioCachingSmokeTest.java | 11 +-
...t.java => TestHudiFlociConnectorSmokeTest.java} | 11 +-
.../hudi/TestHudiMemoryCacheFileOperations.java | 10 +-
.../io/trino/plugin/hudi/TestHudiSmokeTest.java | 51 ++++++-
.../io/trino/plugin/hudi/TestHudiSplitSource.java | 121 +++++++++++++++
.../plugin/hudi/util/TestTupleDomainUtilsTest.java | 2 +-
pom.xml | 11 +-
release/release_guide.md | 27 ++++
rfc/rfc-105/rfc-105.md | 4 +-
scripts/trino/bootstrap_trino.sh | 170 +++++++++++++++++++++
31 files changed, 892 insertions(+), 200 deletions(-)
diff --git a/.github/workflows/hudi_trino_ci.yml
b/.github/workflows/hudi_trino_ci.yml
index 4dad4f1a3cba..dde37ec8a986 100644
--- a/.github/workflows/hudi_trino_ci.yml
+++ b/.github/workflows/hudi_trino_ci.yml
@@ -8,6 +8,7 @@ on:
paths:
- 'hudi-trino/**'
- '.github/workflows/hudi_trino_ci.yml'
+ - 'scripts/trino/**'
# Upstream modules the connector build installs (the -am closure of the
JDK 17
# install step) plus the poms that own trino.version and the dependency
pins.
# Keep in sync with the case patterns in detect-trino-changes below.
@@ -77,6 +78,7 @@ jobs:
case "$f" in
hudi-trino/*) TRINO=true ;;
.github/workflows/hudi_trino_ci.yml) TRINO=true ;;
+ scripts/trino/*) TRINO=true ;;
# Upstream modules the connector build installs (the -am closure
of the JDK 17
# install step) plus the poms that own trino.version and the
dependency pins.
# Keep in sync with the push paths above.
@@ -105,6 +107,21 @@ jobs:
- name: Checkout repository
if: needs.changes.outputs.trino == 'true'
uses: actions/checkout@v5
+ - name: Read Trino pin
+ id: trino-pin
+ if: needs.changes.outputs.trino == 'true'
+ run: |
+ set -euo pipefail
+ TRINO_SHA=$(sed -n 's|.*<trino.sha>\(.*\)</trino.sha>.*|\1|p'
pom.xml)
+ TRINO_VERSION=$(sed -n
's|.*<trino.version>\(.*\)</trino.version>.*|\1|p' pom.xml)
+ # sed -n ...p exits 0 on no match; an empty value would
checkout/cache garbage.
+ if [ -z "$TRINO_SHA" ] || [ -z "$TRINO_VERSION" ]; then
+ echo "ERROR: could not read trino.sha/trino.version from pom.xml"
>&2
+ exit 1
+ fi
+ echo "Pinned trinodb/trino $TRINO_VERSION at $TRINO_SHA"
+ echo "trino_sha=$TRINO_SHA" >> "$GITHUB_OUTPUT"
+ echo "trino_version=$TRINO_VERSION" >> "$GITHUB_OUTPUT"
# Hudi targets Java 11 and uses Lombok 1.18.36, which does not run on
JDK 25.
# Build the upstream modules hudi-trino depends on under JDK 17 first,
# install them into the local m2, then build the connector itself under
JDK 25.
@@ -127,20 +144,32 @@ jobs:
java-version: '25'
distribution: 'temurin'
cache: maven
- # Trino does not publish trino-spi / trino-filesystem / trino-hive
test-jars to
- # Maven Central. Check out the matching release tag and install just the
modules
- # whose test classifiers we need into the local m2.
- - name: Checkout trinodb/trino at 481
+ - name: Purge Trino artifacts from the local m2
if: needs.changes.outputs.trino == 'true'
+ # Artifacts an older pin left behind carry the same SNAPSHOT
coordinates as the current ones.
+ run: rm -rf ~/.m2/repository/io/trino
+ # Trino publishes neither SNAPSHOT artifacts nor the trino-spi /
trino-filesystem /
+ # trino-hive / trino-main test-jars, so every io.trino dependency is
built from the
+ # pinned trinodb/trino commit and cached under that commit.
+ - name: Restore Trino artifacts for the pinned commit
+ id: trino-m2
+ if: needs.changes.outputs.trino == 'true'
+ uses: actions/cache@v4
+ with:
+ path: ~/.m2/repository/io/trino
+ key: trino-m2-v2-${{ hashFiles('scripts/trino/bootstrap_trino.sh')
}}-${{ steps.trino-pin.outputs.trino_sha }}
+ # No restore-keys on purpose: a partial restore from another pin
collides on the
+ # same SNAPSHOT coordinates and poisons the build.
+ - name: Checkout trinodb/trino at the pinned commit
+ if: needs.changes.outputs.trino == 'true' &&
steps.trino-m2.outputs.cache-hit != 'true'
uses: actions/checkout@v5
with:
repository: trinodb/trino
- ref: '481'
+ ref: ${{ steps.trino-pin.outputs.trino_sha }}
path: trino-src
- - name: Install Trino test-jars (JDK 25)
- if: needs.changes.outputs.trino == 'true'
- working-directory: trino-src
- run: mvn $MVN_ARGS install -pl
:trino-spi,:trino-filesystem,:trino-hive,:trino-main -am -DskipTests
-Dair.check.skip-all=true
+ - name: Build Trino artifacts from source (JDK 25)
+ if: needs.changes.outputs.trino == 'true' &&
steps.trino-m2.outputs.cache-hit != 'true'
+ run: scripts/trino/bootstrap_trino.sh trino-src --skip-checkout
- name: Build connector (JDK 25)
if: needs.changes.outputs.trino == 'true'
run: mvn $MVN_ARGS -Phudi-trino -pl hudi-trino install
-Dmaven.test.skip=true
diff --git a/.github/workflows/hudi_trino_compat.yml
b/.github/workflows/hudi_trino_compat.yml
index 701f72115510..eced07275dd5 100644
--- a/.github/workflows/hudi_trino_compat.yml
+++ b/.github/workflows/hudi_trino_compat.yml
@@ -4,11 +4,20 @@ on:
schedule:
- cron: '17 4 * * *'
workflow_dispatch:
+ inputs:
+ trino_ref:
+ description: 'trinodb/trino ref to verify and pin (default: master
HEAD); the release pin-back dispatches this with the released tag'
+ required: false
+ default: 'master'
-# The failure handler files/updates a drift report issue.
+# Two jobs on purpose: the build job executes mvnw from a fresh trinodb/trino
master
+# checkout, so it must never hold a push-capable token; the propose job holds
+# contents: write but runs no third-party code. The failure handler
files/updates a
+# drift report issue; on success the propose job pushes the pin-advance branch
and a
+# human opens the PR (a GITHUB_TOKEN-created PR would start no workflow runs,
so the
+# required checks could never report).
permissions:
contents: read
- issues: write
env:
MVN_ARGS: -e -ntp -B -V -Dgpg.skip -Djacoco.skip
@@ -17,17 +26,31 @@ jobs:
compile-against-trino-master:
name: Compile hudi-trino against trinodb/trino master
runs-on: ubuntu-latest
+ permissions:
+ contents: read
+ issues: write
+ outputs:
+ head_sha: ${{ steps.trino-head.outputs.head_sha }}
+ trino_version: ${{ steps.trino-version.outputs.trino_version }}
steps:
- name: Checkout Hudi
uses: actions/checkout@v5
with:
path: hudi
- - name: Checkout trinodb/trino master
+ - name: Checkout trinodb/trino
uses: actions/checkout@v5
with:
repository: trinodb/trino
- ref: master
+ # Scheduled runs track master; a manual dispatch may pin any ref,
e.g. a release tag
+ ref: ${{ inputs.trino_ref || 'master' }}
path: trino
+ - name: Read trinodb/trino HEAD
+ id: trino-head
+ run: |
+ set -euo pipefail
+ HEAD_SHA=$(git -C trino rev-parse HEAD)
+ echo "trinodb/trino master is at $HEAD_SHA"
+ echo "head_sha=$HEAD_SHA" >> "$GITHUB_OUTPUT"
# Hudi targets Java 11 and uses Lombok 1.18.36, which does not run on
JDK 25.
# Install the upstream Hudi modules under JDK 17 first, then compile the
connector
# under JDK 25.
@@ -39,7 +62,9 @@ jobs:
cache: maven
- name: Install upstream Hudi modules (JDK 17)
working-directory: hudi
- run: mvn $MVN_ARGS install -pl
:hudi-common,:hudi-hive-sync,:hudi-io,:hudi-sync-common -am
-Dmaven.test.skip=true -Drat.skip -Dcheckstyle.skip
+ # hudi-client-common and hudi-java-client are needed by the test step's
+ # hudi-trino-tests profile, same as in hudi_trino_ci.yml.
+ run: mvn $MVN_ARGS install -pl
:hudi-common,:hudi-hive-sync,:hudi-io,:hudi-sync-common,:hudi-client-common,:hudi-java-client
-am -Dmaven.test.skip=true -Drat.skip -Dcheckstyle.skip
- name: Set up JDK 25
uses: actions/setup-java@v5
with:
@@ -59,11 +84,12 @@ jobs:
echo "trino_version=$VERSION" >> "$GITHUB_OUTPUT"
echo "Detected Trino version: $VERSION"
- name: Install Trino modules from master (JDK 25)
- working-directory: trino
- # hudi-trino compiles against these plus their transitive modules
(spi, cache, metastore,
- # hive-formats, memory-context). They must come from the master
checkout -- resolving from
- # Maven Central would defeat the point of the drift check.
- run: mvn $MVN_ARGS install -pl
:trino-hive,:trino-filesystem-manager,:trino-parquet,:trino-plugin-toolkit -am
-DskipTests -Dair.check.skip-all=true
+ # Same module set the pinned build uses, so a green compile here is a
promotable pin. The
+ # script purges ~/.m2/repository/io/trino itself and only warns when
master's version has
+ # rolled past the pinned trino.version.
+ env:
+ HEAD_SHA: ${{ steps.trino-head.outputs.head_sha }}
+ run: hudi/scripts/trino/bootstrap_trino.sh trino --skip-checkout --ref
"$HEAD_SHA"
- name: Compile hudi-trino against current Trino SPI (JDK 25)
id: compile
working-directory: hudi
@@ -71,11 +97,28 @@ jobs:
mvn $MVN_ARGS -Phudi-trino \
-Dtrino.version=${{ steps.trino-version.outputs.trino_version }} \
-pl hudi-trino compile
+ - name: Run hudi-trino tests against current Trino SPI (JDK 25)
+ id: test
+ working-directory: hudi
+ # Drift that only breaks test sources or test behavior (deleted test
containers,
+ # renamed tracing spans) must block the pin advance too, not just
main-source drift.
+ run: |
+ mvn $MVN_ARGS -Phudi-trino,hudi-trino-tests \
+ -Dtrino.version=${{ steps.trino-version.outputs.trino_version }} \
+ -pl hudi-trino test
+ - name: Save Trino artifacts under the candidate pin
+ # Pre-seeds the gating CI cache, which keys on the sha, so the pin
advance below does not
+ # make every PR rebuild Trino from source.
+ if: steps.compile.outcome == 'success'
+ uses: actions/cache/save@v4
+ with:
+ path: ~/.m2/repository/io/trino
+ key: trino-m2-v2-${{
hashFiles('hudi/scripts/trino/bootstrap_trino.sh') }}-${{
steps.trino-head.outputs.head_sha }}
- name: Open issue on failure
- # Only a connector compile failure is SPI drift. A bare failure()
would also file
- # the issue for a failed checkout or a broken trinodb/trino master
build, with an
+ # Only a connector compile or test failure is SPI drift. A bare
failure() would also
+ # file the issue for a failed checkout or a broken trinodb/trino
master build, with an
# empty version in the title when the failure is before Read Trino
version.
- if: failure() && steps.compile.outcome == 'failure'
+ if: failure() && (steps.compile.outcome == 'failure' ||
steps.test.outcome == 'failure')
uses: actions/github-script@v7
with:
script: |
@@ -103,3 +146,59 @@ jobs:
title: `hudi-trino SPI drift detected against Trino ${version}`,
body: `${marker}\nNightly compatibility build failed against
Trino ${version}. See ${runUrl}.`,
});
+
+ propose-pin-advance:
+ name: Push pin-advance branch
+ runs-on: ubuntu-latest
+ needs: compile-against-trino-master
+ # The only job with a push credential; it checks out apache/hudi alone and
runs no
+ # Maven or third-party code. Pin PRs are opened and merged by humans.
+ permissions:
+ contents: write
+ steps:
+ - name: Checkout Hudi
+ uses: actions/checkout@v5
+ - name: Push pin-advance branch
+ env:
+ GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+ HEAD_SHA: ${{ needs.compile-against-trino-master.outputs.head_sha }}
+ TRINO_VERSION: ${{
needs.compile-against-trino-master.outputs.trino_version }}
+ BASE_BRANCH: ${{ github.ref_name }}
+ run: |
+ set -euo pipefail
+ PINNED_VERSION=$(sed -n
's|.*<trino.version>\(.*\)</trino.version>.*|\1|p' pom.xml)
+ git config user.name 'github-actions[bot]'
+ git config user.email
'41898282+github-actions[bot]@users.noreply.github.com'
+ git checkout -B bot/trino-pin
+ sed -i
"s|<trino.sha>.*</trino.sha>|<trino.sha>${HEAD_SHA}</trino.sha>|" pom.xml
+ if [ "$TRINO_VERSION" != "$PINNED_VERSION" ]; then
+ # Version rollover: the shim's parent tracks trino.version.
+ sed -i
"s|<trino.version>.*</trino.version>|<trino.version>${TRINO_VERSION}</trino.version>|"
pom.xml
+ sed -i "/<parent>/,/<\/parent>/
s|<version>.*</version>|<version>${TRINO_VERSION}</version>|"
docker/trino/shim/pom.xml
+ fi
+ # Keep trino.e2e.version current too; without this it silently stops
being
+ # "the latest released Trino" at the next upstream release.
Best-effort: an
+ # API hiccup must not block the pin advance.
+ LATEST_RELEASE=$(gh api repos/trinodb/trino/releases/latest --jq
.tag_name || true)
+ if printf '%s' "$LATEST_RELEASE" | grep -qE '^[0-9]+$'; then
+ sed -i
"s|<trino.e2e.version>.*</trino.e2e.version>|<trino.e2e.version>${LATEST_RELEASE}</trino.e2e.version>|"
pom.xml
+ else
+ echo "Could not determine the latest Trino release
('${LATEST_RELEASE}'); leaving trino.e2e.version unchanged."
+ fi
+ if git diff --quiet; then
+ echo "Pin is already at ${HEAD_SHA}; nothing to propose."
+ exit 0
+ fi
+ TITLE="chore(trino): advance trino master pin to ${HEAD_SHA:0:12}"
+ git commit -am "$TITLE"
+ git push --force origin bot/trino-pin
+ {
+ echo "## Pin advance pushed to bot/trino-pin"
+ echo ""
+ echo "hudi-trino compiles against trinodb/trino ${TRINO_VERSION}
at ${HEAD_SHA}."
+ echo "Open the PR (bot-created PRs trigger no workflow runs, so a
human does this):"
+ echo ""
+ echo '```'
+ echo "gh pr create --repo ${GITHUB_REPOSITORY} --base
${BASE_BRANCH} --head bot/trino-pin --title \"${TITLE}\" --body
\"Nightly-verified pin advance; see the Hudi Trino SPI Compatibility run.\""
+ echo '```'
+ } >> "$GITHUB_STEP_SUMMARY"
diff --git a/.github/workflows/hudi_trino_e2e.yml
b/.github/workflows/hudi_trino_e2e.yml
index df2f945e36d3..e6269c94bd47 100644
--- a/.github/workflows/hudi_trino_e2e.yml
+++ b/.github/workflows/hudi_trino_e2e.yml
@@ -16,9 +16,9 @@ on:
# which compiles the whole module, so a break
# anywhere in it fails this pipeline -- not just
# under integ2/.
- # pom.xml owns trino.version, which the shim pom's
parent,
- # the Dockerfile TRINO_VERSION arg and the
- # hardcoded 481 paths below all track by hand.
+ # pom.xml owns trino.version / trino.sha /
trino.e2e.version:
+ # the shim pom's parent, the plugin dir path and
the
+ # server image version below are all derived
from them.
- 'hudi-trino/**'
- 'docker/trino/**'
- 'docker/compose/docker-compose_hadoop340_hive2310_spark402*'
@@ -27,6 +27,7 @@ on:
- 'hudi-integ-test/**'
- 'pom.xml'
- '.github/workflows/hudi_trino_e2e.yml'
+ - 'scripts/trino/**'
pull_request:
branches:
- master
@@ -41,6 +42,7 @@ on:
- 'hudi-integ-test/**'
- 'pom.xml'
- '.github/workflows/hudi_trino_e2e.yml'
+ - 'scripts/trino/**'
workflow_dispatch:
concurrency:
@@ -62,12 +64,81 @@ jobs:
# hudi-trino at HEAD, assembles the plugin dir via the in-repo shim
# (docker/trino/shim, standing in for the not-yet-released upstream
# trinodb/trino plugin/trino-hudi shim), bakes it into a local
- # apachehudi/hudi-trino_481 image, and runs ITTestTrino* against the
- # spark402 compose stack (the only pair with the trinocoordinator service).
+ # apachehudi/hudi-trino-e2e image on top of the released trino.e2e.version
+ # server, and runs ITTestTrino* against the spark402 compose stack (the
only
+ # pair with the trinocoordinator service).
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v5
+ - name: Read Trino pin
+ id: trino-pin
+ run: |
+ set -euo pipefail
+ TRINO_SHA=$(sed -n 's|.*<trino.sha>\(.*\)</trino.sha>.*|\1|p'
pom.xml)
+ TRINO_VERSION=$(sed -n
's|.*<trino.version>\(.*\)</trino.version>.*|\1|p' pom.xml)
+ E2E_VERSION=$(sed -n
's|.*<trino.e2e.version>\(.*\)</trino.e2e.version>.*|\1|p' pom.xml)
+ # sed -n ...p exits 0 on no match; an empty sha would make the
commits API answer
+ # for the default branch and silently skip the suite, so fail loudly
instead.
+ if [ -z "$TRINO_SHA" ] || [ -z "$TRINO_VERSION" ] || [ -z
"$E2E_VERSION" ]; then
+ echo "ERROR: could not read
trino.sha/trino.version/trino.e2e.version from pom.xml" >&2
+ exit 1
+ fi
+ echo "Connector builds at $TRINO_VERSION ($TRINO_SHA); server image
is $E2E_VERSION"
+ echo "trino_sha=$TRINO_SHA" >> "$GITHUB_OUTPUT"
+ echo "trino_version=$TRINO_VERSION" >> "$GITHUB_OUTPUT"
+ echo "e2e_version=$E2E_VERSION" >> "$GITHUB_OUTPUT"
+ - name: SPI drift gate
+ id: spi-drift
+ # The plugin is built at the pin but loaded by the released
trino.e2e.version server, so
+ # any SPI / filesystem change between the two can make the image
unbootable. Skip the run
+ # instead of reporting a failure that no connector change caused.
+ env:
+ GH_TOKEN: ${{ github.token }}
+ TRINO_SHA: ${{ steps.trino-pin.outputs.trino_sha }}
+ TRINO_VERSION: ${{ steps.trino-pin.outputs.trino_version }}
+ E2E_VERSION: ${{ steps.trino-pin.outputs.e2e_version }}
+ run: |
+ set -euo pipefail
+ # Per-path commit queries, NOT the compare API: compare caps its
file list at 300
+ # and a single Trino release cycle already exceeds that, so a capped
compare would
+ # flag every pin more than a release old as drifted. The commits API
is uncapped;
+ # any commit reachable from the pin that touched a boundary-crossing
path after the
+ # released tag's commit date (excluding the tag commit itself) is
drift. Existence
+ # is enough, so the first page settles it -- truncation cannot yield
a false pass.
+ # Gated paths are the surfaces where server-built and pin-built
classes meet:
+ # core/trino-spi (the server provides it to the plugin classloader),
and the
+ # HdfsFileSystemLoader contract, where bundled
trino-filesystem-manager loads the
+ # server image's version-matched hdfs jar set (see
docker/trino/Dockerfile) whose
+ # HdfsClassLoader then delegates the exact packages
io.trino.filesystem and
+ # io.trino.memory.context back to the plugin's bundled copies -- so
those two libs
+ # cross the boundary precisely BECAUSE they are bundled.
+ TAG_SHA=$(gh api "repos/trinodb/trino/commits/${E2E_VERSION}" --jq
.sha)
+ TAG_DATE=$(gh api "repos/trinodb/trino/commits/${E2E_VERSION}" --jq
.commit.committer.date)
+ DRIFTED=false
+ for p in core/trino-spi lib/trino-filesystem
lib/trino-filesystem-manager lib/trino-hdfs lib/trino-memory-context; do
+ # Assign before iterating: a failing substitution in the for-list
would not trip
+ # set -e, and gh api prints the error body to stdout, so a
transient API error
+ # would otherwise iterate over error JSON and silently skip the
suite.
+ SHAS=$(gh api
"repos/trinodb/trino/commits?sha=${TRINO_SHA}&path=${p}&since=${TAG_DATE}" --jq
'.[].sha')
+ for c in $SHAS; do
+ if [ "$c" != "$TAG_SHA" ]; then
+ echo "Boundary-crossing change under ${p}: ${c}"
+ DRIFTED=true
+ fi
+ done
+ done
+ echo "drifted=$DRIFTED" >> "$GITHUB_OUTPUT"
+ if [ "$DRIFTED" = "true" ]; then
+ {
+ echo "## Trino E2E skipped: SPI drift window"
+ echo ""
+ echo "The connector is built against trinodb/trino
\`${TRINO_VERSION}\` (\`${TRINO_SHA}\`), while the"
+ echo "e2e server image is released Trino \`${E2E_VERSION}\`. SPI
/ filesystem paths changed between"
+ echo "the two, so this run is skipped until the pin and the
released version re-align."
+ } >> "$GITHUB_STEP_SUMMARY"
+ fi
- name: Free disk space
+ if: steps.spi-drift.outputs.drifted != 'true'
run: |
sudo rm -rf /usr/share/dotnet
sudo rm -rf /usr/local/lib/android
@@ -75,12 +146,14 @@ jobs:
sudo rm -rf /usr/local/share/boost
docker system prune --all --force --volumes
- name: Pre-pull compose images (fails fast if not published)
+ if: steps.spi-drift.outputs.drifted != 'true'
run: |
# Surface a missing sparkadhoc image before the long Maven install.
The
# remaining stack images are pulled by docker-compose at test time;
the
# trino image is built locally below, never pulled.
docker pull
apachehudi/hudi-hadoop_3.4.0-hive_2.3.10-sparkadhoc_4.0.2:latest
- name: Set up JDK 17
+ if: steps.spi-drift.outputs.drifted != 'true'
uses: actions/setup-java@v5
with:
java-version: '17'
@@ -88,24 +161,50 @@ jobs:
architecture: x64
cache: maven
- name: Build and install Hudi artifacts (JDK 17)
+ if: steps.spi-drift.outputs.drifted != 'true'
# Full reactor: the compose containers mount the workspace and the
tests
# use bundles staged by the -Pintegration-tests build (e.g.
# docker/hoodie/hadoop/hive_base/target/hoodie-spark-bundle.jar).
run:
mvn clean install -T 2 $SCALA_PROFILE -Dspark4.0 -Dflink1.20
-Pintegration-tests -DskipTests=true -Ddocker.compose.skip=true $MVN_ARGS
- name: Set up JDK 25
+ if: steps.spi-drift.outputs.drifted != 'true'
uses: actions/setup-java@v5
with:
java-version: '25'
distribution: 'temurin'
cache: maven
+ - name: Purge Trino artifacts from the local m2
+ if: steps.spi-drift.outputs.drifted != 'true'
+ # Artifacts an older pin left behind carry the same SNAPSHOT
coordinates as the current ones.
+ run: rm -rf ~/.m2/repository/io/trino
+ # The connector and the shim assembly resolve io.trino from the pinned
trinodb/trino
+ # commit: Trino publishes no SNAPSHOT artifacts, so nothing here comes
from Central.
+ - name: Restore Trino artifacts for the pinned commit
+ id: trino-m2
+ if: steps.spi-drift.outputs.drifted != 'true'
+ uses: actions/cache@v4
+ with:
+ path: ~/.m2/repository/io/trino
+ key: trino-m2-v2-${{ hashFiles('scripts/trino/bootstrap_trino.sh')
}}-${{ steps.trino-pin.outputs.trino_sha }}
+ # No restore-keys on purpose: a partial restore from another pin
collides on the
+ # same SNAPSHOT coordinates and poisons the build.
+ - name: Checkout trinodb/trino at the pinned commit
+ if: steps.spi-drift.outputs.drifted != 'true' &&
steps.trino-m2.outputs.cache-hit != 'true'
+ uses: actions/checkout@v5
+ with:
+ repository: trinodb/trino
+ ref: ${{ steps.trino-pin.outputs.trino_sha }}
+ path: trino-src
+ - name: Build Trino artifacts from source (JDK 25)
+ if: steps.spi-drift.outputs.drifted != 'true' &&
steps.trino-m2.outputs.cache-hit != 'true'
+ run: scripts/trino/bootstrap_trino.sh trino-src --skip-checkout
- name: Build hudi-trino connector (JDK 25)
- # No trinodb/trino checkout needed: the unpublished Trino test-jars sit
- # behind the off-by-default hudi-trino-tests profile and packaging
- # resolves entirely from Maven Central.
+ if: steps.spi-drift.outputs.drifted != 'true'
run:
mvn $MVN_ARGS -Phudi-trino -pl hudi-trino install
-Dmaven.test.skip=true
- name: Assemble trino-hudi plugin dir via in-repo shim (JDK 25)
+ if: steps.spi-drift.outputs.drifted != 'true'
# package, NOT install: installing would shadow the real
# io.trino:trino-hudi release coordinates in the local m2 (the shim pom
# also hard-disables install via maven.install.skip).
@@ -117,20 +216,30 @@ jobs:
HUDI_VERSION=$(mvn -q -ntp help:evaluate
-Dexpression=project.version -DforceStdout)
echo "Building shim against hudi version: $HUDI_VERSION"
mvn $MVN_ARGS -f docker/trino/shim/pom.xml clean package -DskipTests
-Ddep.hudi.version="$HUDI_VERSION"
- - name: Build apachehudi/hudi-trino_481 image
+ - name: Build apachehudi/hudi-trino-e2e image
+ if: steps.spi-drift.outputs.drifted != 'true'
+ # The plugin dir is named after the version it was built at; the
server underneath is
+ # the released trino.e2e.version.
+ env:
+ TRINO_VERSION: ${{ steps.trino-pin.outputs.trino_version }}
+ E2E_VERSION: ${{ steps.trino-pin.outputs.e2e_version }}
run: |
- docker/trino/build_image.sh --plugin-dir
docker/trino/shim/target/trino-hudi-481
+ PLUGIN_DIR="docker/trino/shim/target/trino-hudi-${TRINO_VERSION}"
+ # trino-maven-plugin 24 emits only the zip; explode it into the
plugin dir layout.
+ unzip -o -q "${PLUGIN_DIR}.zip" -d docker/trino/shim/target
+ docker/trino/build_image.sh --plugin-dir "$PLUGIN_DIR"
--trino-version "${E2E_VERSION}"
# Sanity: the shim must have produced a populated plugin dir with a
# service descriptor jar, or Trino cannot load the plugin at boot.
- echo "plugin dir jar count: $(ls
docker/trino/shim/target/trino-hudi-481 | wc -l)"
- ls docker/trino/shim/target/trino-hudi-481/*services*.jar
+ echo "plugin dir jar count: $(ls "$PLUGIN_DIR" | wc -l)"
+ ls "$PLUGIN_DIR"/*services*.jar
- name: Smoke-boot the Trino image standalone
+ if: steps.spi-drift.outputs.drifted != 'true'
# Catches image-level boot failures (plugin load errors, bad etc/
config)
# ~30 min before the IT step would, with the full boot log on screen.
# --hostname trinocoordinator makes the baked discovery.uri
self-resolve.
run: |
docker run -d --name trino-smoke --hostname trinocoordinator \
- apachehudi/hudi-trino_481:latest
+ apachehudi/hudi-trino-e2e:latest
ok=""
for i in $(seq 1 18); do
if [ "$(docker inspect -f '{{.State.Running}}' trino-smoke)" !=
"true" ]; then
@@ -151,6 +260,7 @@ jobs:
fi
docker rm -f trino-smoke
- name: Set up JDK 17 (restore for the IT run)
+ if: steps.spi-drift.outputs.drifted != 'true'
# setup-java resets JAVA_HOME on each call; hudi-integ-test needs 17.
uses: actions/setup-java@v5
with:
@@ -158,6 +268,7 @@ jobs:
distribution: 'temurin'
architecture: x64
- name: Run Trino E2E ITs (JDK 17)
+ if: steps.spi-drift.outputs.drifted != 'true'
run: |
# -DskipITs=false overrides the spark4.0 profile's skipITs=true
default
# (see root pom.xml). -Dcompose.profiles=trino starts the
profile-gated
diff --git a/docker/README.md b/docker/README.md
index 8892399fb33f..560d800063e6 100644
--- a/docker/README.md
+++ b/docker/README.md
@@ -200,9 +200,10 @@ changes are needed for the current amd64 plus arm64 image
set in this repository
## Trino E2E image - `/trino`
The Trino E2E stack does not use the `hoodie/hadoop` image tree.
`docker/trino/` builds
-`apachehudi/hudi-trino_<trino-version>` directly on top of the official
`trinodb/trino`
-image, baking in a locally-assembled native `trino-hudi` plugin directory and
the E2E
-catalog config (`connector.name=hudi`, metastore at
`thrift://hivemetastore:9083`).
+`apachehudi/hudi-trino-e2e` directly on top of the official `trinodb/trino`
image at the
+root pom's `trino.e2e.version`, baking in a locally-assembled native
`trino-hudi` plugin
+directory and the E2E catalog config (`connector.name=hudi`, metastore at
+`thrift://hivemetastore:9083`).
This image is built locally on demand (also by the `hudi_trino_e2e.yml` CI
workflow) and
is NOT published to Docker Hub. The plugin directory comes from the in-repo
shim project
@@ -214,7 +215,9 @@ at `docker/trino/shim/` (see `hudi-trino/README.md` for the
full build-and-run f
# cut_release_branch.sh cannot bump the literal default in its own pom.
HUDI_VERSION=$(mvn -q -ntp help:evaluate -Dexpression=project.version
-DforceStdout)
mvn -f docker/trino/shim/pom.xml clean package -DskipTests
-Ddep.hudi.version="$HUDI_VERSION"
-docker/trino/build_image.sh --plugin-dir
docker/trino/shim/target/trino-hudi-481
+TRINO_VERSION=$(sed -n 's|.*<trino.version>\(.*\)</trino.version>.*|\1|p'
pom.xml)
+unzip -o -q "docker/trino/shim/target/trino-hudi-$TRINO_VERSION.zip" -d
docker/trino/shim/target # trino-maven-plugin 24 emits only the zip
+docker/trino/build_image.sh --plugin-dir
"docker/trino/shim/target/trino-hudi-$TRINO_VERSION"
```
The `trinocoordinator` compose service exists only in the
diff --git
a/docker/compose/docker-compose_hadoop340_hive2310_spark402_amd64.yml
b/docker/compose/docker-compose_hadoop340_hive2310_spark402_amd64.yml
index adf964e2288b..a15ea8bfcccb 100644
--- a/docker/compose/docker-compose_hadoop340_hive2310_spark402_amd64.yml
+++ b/docker/compose/docker-compose_hadoop340_hive2310_spark402_amd64.yml
@@ -261,7 +261,7 @@ services:
# docker/trino/empty-overlay (baked-in plugin used); set TRINO_PLUGIN_DIR to
a
# locally-built trino-hudi plugin dir to override it at container start.
trinocoordinator:
- image: apachehudi/hudi-trino_481:latest
+ image: apachehudi/hudi-trino-e2e:latest
profiles: ["trino"]
hostname: trinocoordinator
container_name: trinocoordinator
diff --git
a/docker/compose/docker-compose_hadoop340_hive2310_spark402_arm64.yml
b/docker/compose/docker-compose_hadoop340_hive2310_spark402_arm64.yml
index adf964e2288b..a15ea8bfcccb 100644
--- a/docker/compose/docker-compose_hadoop340_hive2310_spark402_arm64.yml
+++ b/docker/compose/docker-compose_hadoop340_hive2310_spark402_arm64.yml
@@ -261,7 +261,7 @@ services:
# docker/trino/empty-overlay (baked-in plugin used); set TRINO_PLUGIN_DIR to
a
# locally-built trino-hudi plugin dir to override it at container start.
trinocoordinator:
- image: apachehudi/hudi-trino_481:latest
+ image: apachehudi/hudi-trino-e2e:latest
profiles: ["trino"]
hostname: trinocoordinator
container_name: trinocoordinator
diff --git a/docker/trino/Dockerfile b/docker/trino/Dockerfile
index 3189368a2283..891a41aa1ec7 100644
--- a/docker/trino/Dockerfile
+++ b/docker/trino/Dockerfile
@@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
-ARG TRINO_VERSION=481
+ARG TRINO_VERSION=483
FROM trinodb/trino:${TRINO_VERSION}
USER root
diff --git a/docker/trino/build_image.sh b/docker/trino/build_image.sh
index da1cc8fe2ef1..9a2303ff7117 100755
--- a/docker/trino/build_image.sh
+++ b/docker/trino/build_image.sh
@@ -15,21 +15,25 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-# Builds the apachehudi/hudi-trino_<version> image with a locally-built
-# trino-hudi plugin baked in. The plugin dir (typically the in-repo shim's
+# Builds the apachehudi/hudi-trino-e2e image with a locally-built trino-hudi
+# plugin baked in. The plugin dir (typically the in-repo shim's
# docker/trino/shim/target/trino-hudi-<v>, see docker/trino/shim/pom.xml) is
# staged into the build context at docker/trino/plugin/ (gitignored), then
# baked into the image.
# Usage: ./build_image.sh --plugin-dir <path> [--trino-version <v>]
[--image-tag <t>]
-# Typical: ./build_image.sh --plugin-dir "$(dirname
"$0")/shim/target/trino-hudi-481"
-# Note: --trino-version must match the shim pom's parent version and the root
-# pom's trino.version property.
+# Typical: ./build_image.sh --plugin-dir "$(dirname
"$0")/shim/target/trino-hudi-<trino.version>"
+# Note: --trino-version is the released Trino server image to build on top of
+# (the root pom's trino.e2e.version), not the version the plugin was built at.
set -e
-# Default values
+# Directory of this script, so the build context and pom lookups are stable
regardless of cwd
+SCRIPT_DIR=$(cd $(dirname "$0") && pwd)
+
+# Default values. The server version defaults to the root pom's
trino.e2e.version (the
+# nightly pin-advance job keeps that current; a literal default here would
rot).
PLUGIN_DIR=""
-TRINO_VERSION="481"
+TRINO_VERSION=$(sed -n
's|.*<trino.e2e.version>\(.*\)</trino.e2e.version>.*|\1|p'
"$SCRIPT_DIR/../../pom.xml")
IMAGE_TAG="latest"
# Parse command-line arguments
@@ -43,8 +47,10 @@ while [[ "$#" -gt 0 ]]; do
shift
done
-# Directory of this script, so the build context path is stable regardless of
cwd
-SCRIPT_DIR=$(cd $(dirname "$0") && pwd)
+if [ -z "$TRINO_VERSION" ]; then
+ echo "Error: could not read trino.e2e.version from the root pom and no
--trino-version given." >&2
+ exit 1
+fi
# Validate --plugin-dir: required, must exist and be non-empty
if [ -z "$PLUGIN_DIR" ]; then
@@ -66,7 +72,7 @@ echo "Staging plugin from '$PLUGIN_DIR' into '$STAGE_DIR'"
rm -rf "$STAGE_DIR"
cp -r "$PLUGIN_DIR" "$STAGE_DIR"
-IMAGE="apachehudi/hudi-trino_${TRINO_VERSION}:${IMAGE_TAG}"
+IMAGE="apachehudi/hudi-trino-e2e:${IMAGE_TAG}"
echo "Building $IMAGE (TRINO_VERSION=${TRINO_VERSION})"
docker build --build-arg TRINO_VERSION="${TRINO_VERSION}" -t "$IMAGE"
"$SCRIPT_DIR"
diff --git a/docker/trino/etc/jvm.config b/docker/trino/etc/jvm.config
index b1d8ff3772dc..8a7098469534 100644
--- a/docker/trino/etc/jvm.config
+++ b/docker/trino/etc/jvm.config
@@ -34,5 +34,5 @@
# Allow loading dynamic agents (used by JOL, referenced by Trino's runtime).
-XX:+EnableDynamicAgentLoading
# NOTE: do NOT add -XX:GCLockerRetryAllocationCount here (Hudi's JDK 11/17 CI
-# workaround): the GCLocker was removed in modern JDKs and the
trinodb/trino:481
+# workaround): the GCLocker was removed in modern JDKs and the trinodb/trino
server
# JVM (JDK 25) refuses to start on the unrecognized option.
diff --git a/docker/trino/shim/pom.xml b/docker/trino/shim/pom.xml
index c263b0d43b21..b614a4194f6b 100644
--- a/docker/trino/shim/pom.xml
+++ b/docker/trino/shim/pom.xml
@@ -33,8 +33,9 @@
installing would shadow the real io.trino:trino-hudi release coordinates in
the local repository.
- The parent version below must stay in sync with trino.version in the root
pom,
- the TRINO_VERSION ARG in docker/trino/Dockerfile, and the compose image tag.
+ The parent version below must stay in sync with trino.version in the root
pom. On
+ master that version is a -SNAPSHOT and resolves from the local m2 only, so
run
+ scripts/trino/bootstrap_trino.sh first; on release branches it resolves from
Central.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
<modelVersion>4.0.0</modelVersion>
@@ -42,8 +43,8 @@
<parent>
<groupId>io.trino</groupId>
<artifactId>trino-root</artifactId>
- <version>481</version>
- <!-- Resolve the parent from Maven Central, never the file system. -->
+ <version>484-SNAPSHOT</version>
+ <!-- Resolve the parent from the repositories, never the file system.
-->
<relativePath />
</parent>
@@ -153,9 +154,10 @@
<scope>provided</scope>
</dependency>
- <!-- NOTE: unlike the upstream 482-SNAPSHOT shim, jts-core must NOT be
- declared provided here: it is not part of the Trino 481 SPI
surface
- and SpiDependencyChecker rejects provided scope for it. It ships
- inside the plugin dir at its transitive scope instead. -->
+ <dependency>
+ <groupId>org.locationtech.jts</groupId>
+ <artifactId>jts-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/hudi-trino/README.md b/hudi-trino/README.md
index 72c848ba2c18..1308a21698b9 100644
--- a/hudi-trino/README.md
+++ b/hudi-trino/README.md
@@ -19,9 +19,21 @@
Hudi connector for Trino (RFC-105). Published as `org.apache.hudi:hudi-trino`
-- a regular non-shaded JAR. The Trino-side `trino-hudi` plugin module depends
on this artifact and Trino's URLClassLoader isolates the plugin's transitive
deps from the rest of the server, so no shading is required.
+## Bootstrap Trino (do this first)
+
+On master no `io.trino` artifact resolves from Maven Central: master tracks
`trinodb/trino` master at the commit pinned by `trino.sha` in the root pom, and
Trino publishes no SNAPSHOT artifacts. Pin ownership: the nightly `Hudi Trino
SPI Compatibility` job verifies trino HEAD and pushes a `bot/trino-pin` branch
(also refreshing `trino.e2e.version`); a committer opens and merges its PR, so
the pin advances roughly nightly-to-weekly. One time per pin advance, clone
`trinodb/trino` (or reus [...]
+
+```
+scripts/trino/bootstrap_trino.sh /path/to/trino
+```
+
+Takes roughly 10-30 minutes and installs everything the connector needs,
including the four test-jars and the `trino-root` pom.
+
+On release branches `trino.version` is a released number, so compile deps
resolve from Central and bootstrap is only needed for running tests.
+
## Build
-Excluded from default builds. Activate the `hudi-trino` Maven profile:
+Excluded from default builds. Activate the `hudi-trino` Maven profile (after
bootstrap):
```
# tests need Trino test-jars not on Maven Central (see Running tests); skip
them in the default build
@@ -32,18 +44,15 @@ Requires JDK 25 (enforced via `maven-enforcer-plugin`).
## Running tests
-Tests depend on Trino test-jars (`trino-spi`, `trino-filesystem`,
`trino-hive`, `trino-main` at the `tests` classifier). Trino does not publish
three of those to Maven Central, so the test deps live behind the
`hudi-trino-tests` profile, off by default.
+Tests depend on Trino test-jars (`trino-spi`, `trino-filesystem`,
`trino-hive`, `trino-main` at the `tests` classifier). Trino publishes none of
those, so the test deps live behind the `hudi-trino-tests` profile, off by
default.
-To run the tests:
-
-1. Build the matching Trino version locally so its `*-tests.jar` artifacts
land in your `~/.m2` (see `trino.version` in the root pom for the version to
build).
-2. Activate both profiles:
+After bootstrap, activate both profiles:
```
mvn -Phudi-trino,hudi-trino-tests -pl hudi-trino test
```
-CI follows the same two steps: `.github/workflows/hudi_trino_ci.yml` installs
the test-jars from a source checkout of the pinned Trino tag, then runs with
both profiles enabled.
+CI follows the same two steps: `.github/workflows/hudi_trino_ci.yml` runs
`bootstrap_trino.sh` against the pinned commit (cached per `trino.sha`), then
runs with both profiles enabled.
## End-to-end tests (docker)
@@ -55,7 +64,11 @@ project mirroring the upstream `trinodb/trino`
`plugin/trino-hudi` shim planned
RFC-105 (not yet released upstream). CI runs the same flow via
`.github/workflows/hudi_trino_e2e.yml`.
-Local flow:
+The plugin is built at the pinned `trino.version` while the server image is
the released
+`trino.e2e.version`; CI auto-skips the suite during SPI drift windows (SPI or
filesystem changes
+between the two).
+
+Local flow (after bootstrap):
```
# 1. JDK 17: full reactor incl. the integ-test bundles the containers mount
@@ -69,11 +82,16 @@ mvn -Phudi-trino -pl hudi-trino install
-Dmaven.test.skip=true
# shadow the real io.trino:trino-hudi release coordinates in the local m2).
# dep.hudi.version comes from the reactor pom: the shim sits outside the
# reactor, so cut_release_branch.sh cannot bump its literal default.
+# The unzip lives here, not in step 4: the fast-iteration loop below repeats
+# steps 2-3 only, and `clean` wipes the previously exploded dir.
HUDI_VERSION=$(mvn -q -ntp help:evaluate -Dexpression=project.version
-DforceStdout)
+TRINO_VERSION=$(sed -n 's|.*<trino.version>\(.*\)</trino.version>.*|\1|p'
pom.xml)
mvn -f docker/trino/shim/pom.xml clean package -DskipTests
-Ddep.hudi.version="$HUDI_VERSION"
+unzip -o -q "docker/trino/shim/target/trino-hudi-$TRINO_VERSION.zip" -d
docker/trino/shim/target # trino-maven-plugin 24 emits only the zip
-# 4. Build the Trino image (locally tagged; never published)
-docker/trino/build_image.sh --plugin-dir
docker/trino/shim/target/trino-hudi-481
+# 4. Build the Trino image (locally tagged; never published). The base server
defaults to
+# trino.e2e.version; pass --trino-version to override it.
+docker/trino/build_image.sh --plugin-dir
"docker/trino/shim/target/trino-hudi-$TRINO_VERSION"
# 5. JDK 17: run the suite (only the spark402 compose pair has the trino
service)
mvn verify -pl hudi-integ-test -Dscala-2.13 -Dscala.binary.version=2.13
-Dspark4.0 \
@@ -83,7 +101,7 @@ mvn verify -pl hudi-integ-test -Dscala-2.13
-Dscala.binary.version=2.13 -Dspark4
```
Fast iteration loop: after changing connector code, redo steps 2-3, then add
-`-Dtrino.plugin.dir=$PWD/docker/trino/shim/target/trino-hudi-481` to step 5.
The
+`-Dtrino.plugin.dir=$PWD/docker/trino/shim/target/trino-hudi-$TRINO_VERSION`
to step 5. The
container's overlay entrypoint swaps the freshly built plugin dir in at start,
so the
image rebuild (step 4) is skipped.
diff --git a/hudi-trino/pom.xml b/hudi-trino/pom.xml
index 4a488ee7309b..d2454ca91978 100644
--- a/hudi-trino/pom.xml
+++ b/hudi-trino/pom.xml
@@ -391,6 +391,12 @@
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.locationtech.jts</groupId>
+ <artifactId>jts-core</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.openjdk.jol</groupId>
<artifactId>jol-core</artifactId>
@@ -445,6 +451,18 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>io.trino</groupId>
+ <artifactId>trino-blob-cache-alluxio</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>io.trino</groupId>
+ <artifactId>trino-blob-cache-memory</artifactId>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>io.trino</groupId>
<artifactId>trino-client</artifactId>
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
index 2803f8ae231a..1d58eb7adc1c 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java
@@ -45,6 +45,7 @@ import io.trino.spi.connector.ConnectorPageSource;
import io.trino.spi.connector.ConnectorPageSourceProvider;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.ConnectorSplit;
+import io.trino.spi.connector.ConnectorTableCredentials;
import io.trino.spi.connector.ConnectorTableHandle;
import io.trino.spi.connector.ConnectorTransactionHandle;
import io.trino.spi.connector.DynamicFilter;
@@ -142,11 +143,13 @@ public class HudiPageSourceProvider
}
@Override
+ @SuppressWarnings("deprecation")
public ConnectorPageSource createPageSource(
ConnectorTransactionHandle transaction,
ConnectorSession session,
ConnectorSplit connectorSplit,
ConnectorTableHandle connectorTable,
+ Optional<ConnectorTableCredentials> tableCredentials,
List<ColumnHandle> columns,
DynamicFilter dynamicFilter)
{
diff --git a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplit.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplit.java
index 32f1d32aa3f6..9159ec060ca2 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplit.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplit.java
@@ -17,7 +17,6 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hive.HivePartitionKey;
import io.trino.plugin.hudi.file.HudiBaseFile;
@@ -28,7 +27,6 @@ import io.trino.spi.connector.ConnectorSplit;
import io.trino.spi.predicate.TupleDomain;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import static com.google.common.base.MoreObjects.toStringHelper;
@@ -80,15 +78,6 @@ public class HudiSplit
this.cachingHostAddresses = requireNonNull(cachingHostAddresses,
"cachingHostAddresses is null");
}
- public Map<String, String> getSplitInfo()
- {
- return ImmutableMap.<String, String>builder()
- .put("baseFile", baseFile.toString())
- .put("logFiles", logFiles.toString())
- .put("commitTime", commitTime)
- .buildOrThrow();
- }
-
@JsonProperty
public Optional<HudiBaseFile> getBaseFile()
{
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
index 7084435b8733..df0018a49268 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitManager.java
@@ -24,13 +24,13 @@ import
io.trino.plugin.base.classloader.ClassLoaderSafeConnectorSplitSource;
import io.trino.plugin.hive.HiveColumnHandle;
import io.trino.plugin.hive.HiveTransactionHandle;
import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ColumnHandle;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.ConnectorSplitManager;
import io.trino.spi.connector.ConnectorSplitSource;
import io.trino.spi.connector.ConnectorTableHandle;
import io.trino.spi.connector.ConnectorTransactionHandle;
import io.trino.spi.connector.Constraint;
-import io.trino.spi.connector.DynamicFilter;
import io.trino.spi.connector.TableNotFoundException;
import io.trino.spi.security.ConnectorIdentity;
import org.apache.hudi.common.util.HoodieTimer;
@@ -39,6 +39,7 @@ import org.apache.hudi.common.util.Lazy;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.BiFunction;
@@ -77,7 +78,7 @@ public class HudiSplitManager
ConnectorTransactionHandle transaction,
ConnectorSession session,
ConnectorTableHandle tableHandle,
- DynamicFilter dynamicFilter,
+ Set<ColumnHandle> dynamicFilterColumns,
Constraint constraint)
{
HudiTableHandle hudiTableHandle = (HudiTableHandle) tableHandle;
@@ -98,7 +99,6 @@ public class HudiSplitManager
getMaxSplitsPerSecond(session),
getMaxOutstandingSplits(session),
lazyAllPartitions,
- dynamicFilter,
getDynamicFilteringWaitTimeout(session));
return new ClassLoaderSafeConnectorSplitSource(splitSource,
HudiSplitManager.class.getClassLoader());
}
diff --git a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
index ab6bae93cfc8..a578cc9b73c9 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/HudiSplitSource.java
@@ -13,7 +13,6 @@
*/
package io.trino.plugin.hudi;
-import com.google.common.base.Stopwatch;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
import io.airlift.log.Logger;
@@ -34,7 +33,8 @@ import io.trino.spi.TrinoException;
import io.trino.spi.connector.ConnectorSession;
import io.trino.spi.connector.ConnectorSplit;
import io.trino.spi.connector.ConnectorSplitSource;
-import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.connector.DynamicFilterSnapshot;
+import io.trino.spi.connector.SchemaTableName;
import io.trino.spi.predicate.Domain;
import io.trino.spi.predicate.NullableValue;
import io.trino.spi.predicate.TupleDomain;
@@ -56,7 +56,10 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
@@ -66,22 +69,18 @@ import static
io.trino.plugin.hudi.HudiSessionProperties.getMinimumAssignedSplit
import static
io.trino.plugin.hudi.HudiSessionProperties.getStandardSplitWeightSize;
import static
io.trino.plugin.hudi.HudiSessionProperties.isHudiMetadataTableEnabled;
import static
io.trino.plugin.hudi.HudiSessionProperties.isSizeBasedSplitWeightsEnabled;
-import static java.util.Objects.requireNonNull;
import static java.util.concurrent.CompletableFuture.completedFuture;
-import static java.util.concurrent.TimeUnit.MILLISECONDS;
public class HudiSplitSource
implements ConnectorSplitSource
{
private static final Logger log = Logger.get(HudiSplitSource.class);
- private static final ConnectorSplitBatch EMPTY_BATCH = new
ConnectorSplitBatch(ImmutableList.of(), false);
private final AsyncQueue<ConnectorSplit> queue;
private final ScheduledFuture splitLoaderFuture;
private final AtomicReference<TrinoException> trinoException = new
AtomicReference<>();
- private final DynamicFilter dynamicFilter;
+ private final AtomicBoolean finished = new AtomicBoolean();
private final long dynamicFilteringWaitTimeoutMillis;
- private final Stopwatch dynamicFilterWaitStopwatch;
public HudiSplitSource(
ConnectorSession session,
@@ -91,77 +90,85 @@ public class HudiSplitSource
int maxSplitsPerSecond,
int maxOutstandingSplits,
Lazy<Map<String, Partition>> lazyPartitions,
- DynamicFilter dynamicFilter,
Duration dynamicFilteringWaitTimeoutMillis)
{
- boolean enableMetadataTable = isHudiMetadataTableEnabled(session);
- Lazy<HoodieTableMetadata> lazyTableMetadata = Lazy.lazily(() -> {
- HoodieTimer timer = HoodieTimer.start();
- HoodieMetadataConfig metadataConfig =
HoodieMetadataConfig.newBuilder()
- .enable(enableMetadataTable)
- .build();
- HoodieTableMetaClient metaClient = tableHandle.getMetaClient();
- HoodieEngineContext engineContext = new
HoodieLocalEngineContext(metaClient.getStorage().getConf());
-
- // Defer to the native factory, which creates a
HoodieBackedTableMetadata when the
- // metadata table is enabled and initialized, and falls back to
FileSystemBackedTableMetadata
- // otherwise.
- HoodieTableMetadata tableMetadata =
NativeTableMetadataFactory.getInstance().create(
- engineContext, metaClient.getStorage(), metadataConfig,
metaClient.getBasePath().toString(), true);
- log.info("Loaded table metadata for table: %s in %s ms",
tableHandle.getSchemaTableName(), timer.endTimer());
- return tableMetadata;
- });
+ this(
+ new ThrottledAsyncQueue<>(maxSplitsPerSecond,
maxOutstandingSplits, executor),
+ splitLoaderExecutorService,
+ (queue, errorListener) -> {
+ boolean enableMetadataTable =
isHudiMetadataTableEnabled(session);
+ Lazy<HoodieTableMetadata> lazyTableMetadata =
Lazy.lazily(() -> {
+ HoodieTimer timer = HoodieTimer.start();
+ HoodieMetadataConfig metadataConfig =
HoodieMetadataConfig.newBuilder()
+ .enable(enableMetadataTable)
+ .build();
+ HoodieTableMetaClient metaClient =
tableHandle.getMetaClient();
+ HoodieEngineContext engineContext = new
HoodieLocalEngineContext(metaClient.getStorage().getConf());
+
+ // Defer to the native factory, which creates a
HoodieBackedTableMetadata when the
+ // metadata table is enabled and initialized, and
falls back to FileSystemBackedTableMetadata
+ // otherwise.
+ HoodieTableMetadata tableMetadata =
NativeTableMetadataFactory.getInstance().create(
+ engineContext, metaClient.getStorage(),
metadataConfig, metaClient.getBasePath().toString(), true);
+ log.info("Loaded table metadata for table: %s in %s
ms", tableHandle.getSchemaTableName(), timer.endTimer());
+ return tableMetadata;
+ });
+
+ HudiDirectoryLister hudiDirectoryLister = new
HudiSnapshotDirectoryLister(
+ session,
+ tableHandle,
+ enableMetadataTable,
+ lazyTableMetadata);
- HudiDirectoryLister hudiDirectoryLister = new
HudiSnapshotDirectoryLister(
- session,
- tableHandle,
- enableMetadataTable,
- lazyTableMetadata);
-
- this.queue = new ThrottledAsyncQueue<>(maxSplitsPerSecond,
maxOutstandingSplits, executor);
- HudiBackgroundSplitLoader splitLoader = new HudiBackgroundSplitLoader(
- session,
- tableHandle,
- hudiDirectoryLister,
- queue,
- executor,
- createSplitWeightProvider(session),
- lazyPartitions,
- enableMetadataTable,
- lazyTableMetadata,
- throwable -> {
- trinoException.compareAndSet(null, new
TrinoException(HUDI_CANNOT_OPEN_SPLIT,
- "Failed to generate splits for " +
tableHandle.getSchemaTableName(), throwable));
- queue.finish();
- });
+ return new HudiBackgroundSplitLoader(
+ session,
+ tableHandle,
+ hudiDirectoryLister,
+ queue,
+ executor,
+ createSplitWeightProvider(session),
+ lazyPartitions,
+ enableMetadataTable,
+ lazyTableMetadata,
+ errorListener);
+ },
+ tableHandle.getSchemaTableName(),
+ dynamicFilteringWaitTimeoutMillis);
+ }
+
+ // Visible for tests: lets a test drive the split loader directly while
keeping the
+ // error-listener wiring (set trinoException, then finish the queue --
isFinished
+ // relies on that order) identical to production.
+ HudiSplitSource(
+ AsyncQueue<ConnectorSplit> queue,
+ ScheduledExecutorService splitLoaderExecutorService,
+ BiFunction<AsyncQueue<ConnectorSplit>, Consumer<Throwable>,
Runnable> splitLoaderFactory,
+ SchemaTableName tableName,
+ Duration dynamicFilteringWaitTimeoutMillis)
+ {
+ this.queue = queue;
+ Runnable splitLoader = splitLoaderFactory.apply(queue, throwable -> {
+ trinoException.compareAndSet(null, new
TrinoException(HUDI_CANNOT_OPEN_SPLIT,
+ "Failed to generate splits for " + tableName, throwable));
+ queue.finish();
+ });
this.splitLoaderFuture =
splitLoaderExecutorService.schedule(splitLoader, 0, TimeUnit.MILLISECONDS);
- this.dynamicFilter = requireNonNull(dynamicFilter, "dynamicFilter is
null");
this.dynamicFilteringWaitTimeoutMillis =
dynamicFilteringWaitTimeoutMillis.toMillis();
- this.dynamicFilterWaitStopwatch = Stopwatch.createStarted();
}
@Override
- public CompletableFuture<ConnectorSplitBatch> getNextBatch(int maxSize)
+ public CompletableFuture<List<ConnectorSplit>> getNextBatch(int maxSize,
DynamicFilterSnapshot dynamicFilterSnapshot)
{
- // If dynamic filtering is enabled and we haven't timed out, wait for
the build side to provide the dynamic filter.
- long timeLeft = dynamicFilteringWaitTimeoutMillis -
dynamicFilterWaitStopwatch.elapsed(MILLISECONDS);
- if (dynamicFilter.isAwaitable() && timeLeft > 0) {
- // If the filter is not ready, return an empty batch. The query
engine will call getNextBatch() again.
- // As long as isFinished() is false, effectively polling until the
filter is ready or timeout occurs.
- return dynamicFilter.isBlocked()
- .thenApply(_ -> EMPTY_BATCH)
- .completeOnTimeout(EMPTY_BATCH, timeLeft, MILLISECONDS);
- }
-
TupleDomain<HiveColumnHandle> dynamicFilterPredicate =
-
dynamicFilter.getCurrentPredicate().transformKeys(HiveColumnHandle.class::cast);
+
dynamicFilterSnapshot.currentPredicate().transformKeys(HiveColumnHandle.class::cast);
if (dynamicFilterPredicate.isNone()) {
close();
- return completedFuture(new ConnectorSplitBatch(ImmutableList.of(),
true));
+ // The queue may still hold undrained splits, so
queue.isFinished() would never turn true here
+ finished.set(true);
+ return completedFuture(ImmutableList.of());
}
- boolean noMoreSplits = isFinished();
Throwable throwable = trinoException.get();
if (throwable != null) {
return CompletableFuture.failedFuture(throwable);
@@ -169,13 +176,9 @@ public class HudiSplitSource
return toCompletableFuture(Futures.transform(
queue.getBatchAsync(maxSize),
- splits ->
- {
- List<ConnectorSplit> filteredSplits = splits.stream()
- .filter(split ->
partitionMatchesPredicate((HudiSplit) split, dynamicFilterPredicate))
- .collect(toImmutableList());
- return new ConnectorSplitBatch(filteredSplits,
noMoreSplits);
- },
+ splits -> splits.stream()
+ .filter(split -> partitionMatchesPredicate((HudiSplit)
split, dynamicFilterPredicate))
+ .collect(toImmutableList()),
directExecutor()));
}
@@ -188,7 +191,17 @@ public class HudiSplitSource
@Override
public boolean isFinished()
{
- return splitLoaderFuture.isDone() && queue.isFinished();
+ // The failure callback sets trinoException before finishing the
queue, so once the queue
+ // reports finished the exception (if any) is visible here. Claiming
finished while an
+ // exception is pending would let the engine stop polling and end the
scan silently;
+ // reporting unfinished instead makes the next getNextBatch surface
the failure.
+ return finished.get() || (splitLoaderFuture.isDone() &&
queue.isFinished() && trinoException.get() == null);
+ }
+
+ @Override
+ public long getRequestedDynamicFilterWaitTimeoutMillis()
+ {
+ return dynamicFilteringWaitTimeoutMillis;
}
public static HudiSplitWeightProvider
createSplitWeightProvider(ConnectorSession session)
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java
index cd07e1ab3f13..0131fa7bbe45 100644
---
a/hudi-trino/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/cache/HudiCacheKeyProvider.java
@@ -15,6 +15,7 @@ package io.trino.plugin.hudi.cache;
import io.trino.filesystem.TrinoInputFile;
import io.trino.filesystem.cache.CacheKeyProvider;
+import io.trino.spi.cache.CacheKey;
import java.util.Optional;
@@ -22,7 +23,7 @@ public class HudiCacheKeyProvider
implements CacheKeyProvider
{
@Override
- public Optional<String> getCacheKey(TrinoInputFile inputFile)
+ public Optional<CacheKey> getCacheKey(TrinoInputFile inputFile)
{
String path = inputFile.location().path();
if (path.endsWith(".trinoSchema") ||
path.contains("/.trinoPermissions/")
@@ -32,7 +33,8 @@ public class HudiCacheKeyProvider
// and Hudi table properties file that can be mutable
return Optional.empty();
}
- // Other Hudi data and metadata files are immutable
- return Optional.of(path);
+ // Other Hudi data and metadata files are immutable, so the location
alone identifies the
+ // contents; keying by its components also lets a directory's entries
be invalidated by prefix
+ return Optional.of(CacheKeyProvider.locationKey(inputFile.location()));
}
}
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
index e33bf772dc1f..ee46dd8fa508 100644
---
a/hudi-trino/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
+++
b/hudi-trino/src/main/java/io/trino/plugin/hudi/split/HudiBackgroundSplitLoader.java
@@ -54,6 +54,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;
@@ -166,6 +167,19 @@ public class HudiBackgroundSplitLoader
Futures.whenAllComplete(futures)
.run(() -> {
+ // Guava does not order the per-future exception callbacks
against this
+ // combiner, so a failure in the last future could
otherwise finish the
+ // queue before the error listener records it and
HudiSplitSource.isFinished
+ // would answer true with no pending exception. Surface
failures here first;
+ // the listener is idempotent, so double reporting is
harmless.
+ for (ListenableFuture<Void> future : futures) {
+ try {
+ Futures.getDone(future);
+ }
+ catch (ExecutionException e) {
+ errorListener.accept(e.getCause());
+ }
+ }
asyncQueue.finish();
log.info("Partition pruning split generation finished on
table %s.%s", tableHandle.getSchemaName(), tableHandle.getTableName());
}, directExecutor());
diff --git
a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
index 8b035c084af6..cb2414ee40d8 100644
--- a/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
+++ b/hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiAvroSerializer.java
@@ -496,7 +496,7 @@ public class HudiAvroSerializer
type.writeObject(output, trinoNativeDecimalValue);
}
else {
- throw new TrinoException(GENERIC_INTERNAL_ERROR, "Unhandled type
for Object: " + type.getTypeSignature());
+ throw new TrinoException(GENERIC_INTERNAL_ERROR, "Unhandled type
for Object: " + type.getTypeDescriptor());
}
}
diff --git a/hudi-trino/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
index e11efbf3a569..2939b87384b0 100644
--- a/hudi-trino/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
+++ b/hudi-trino/src/test/java/io/trino/plugin/hudi/HudiQueryRunner.java
@@ -17,11 +17,12 @@ import
com.google.errorprone.annotations.CanIgnoreReturnValue;
import io.airlift.log.Level;
import io.airlift.log.Logger;
import io.airlift.log.Logging;
+import io.trino.blob.cache.memory.MemoryBlobCachePlugin;
import io.trino.filesystem.Location;
import io.trino.metastore.Database;
import io.trino.metastore.HiveMetastoreFactory;
import io.trino.plugin.base.util.Closables;
-import io.trino.plugin.hive.containers.Hive3MinioDataLake;
+import io.trino.plugin.hive.containers.Hive3FlociDataLake;
import io.trino.plugin.hudi.testing.HudiTablesInitializer;
import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
import io.trino.plugin.hudi.testing.TpchHudiTablesInitializer;
@@ -35,9 +36,9 @@ import java.util.Map;
import java.util.Optional;
import static io.trino.testing.TestingSession.testSessionBuilder;
-import static io.trino.testing.containers.Minio.MINIO_ROOT_USER;
-import static io.trino.testing.containers.Minio.MINIO_REGION;
-import static io.trino.testing.containers.Minio.MINIO_ROOT_PASSWORD;
+import static io.trino.testing.containers.Floci.FLOCI_ACCESS_KEY;
+import static io.trino.testing.containers.Floci.FLOCI_REGION;
+import static io.trino.testing.containers.Floci.FLOCI_SECRET_KEY;
import static java.util.Objects.requireNonNull;
public final class HudiQueryRunner
@@ -56,14 +57,14 @@ public final class HudiQueryRunner
return new Builder("local:///");
}
- public static Builder builder(Hive3MinioDataLake hiveMinioDataLake)
+ public static Builder builder(Hive3FlociDataLake hiveFlociDataLake)
{
- return new Builder("s3://" + hiveMinioDataLake.getBucketName() + "/")
- .addConnectorProperty("fs.native-s3.enabled", "true")
- .addConnectorProperty("s3.aws-access-key", MINIO_ROOT_USER)
- .addConnectorProperty("s3.aws-secret-key", MINIO_ROOT_PASSWORD)
- .addConnectorProperty("s3.region", MINIO_REGION)
- .addConnectorProperty("s3.endpoint",
hiveMinioDataLake.getMinio().getMinioAddress())
+ return new Builder("s3://" + hiveFlociDataLake.getBucketName() + "/")
+ .addConnectorProperty("fs.s3.enabled", "true")
+ .addConnectorProperty("s3.aws-access-key", FLOCI_ACCESS_KEY)
+ .addConnectorProperty("s3.aws-secret-key", FLOCI_SECRET_KEY)
+ .addConnectorProperty("s3.region", FLOCI_REGION)
+ .addConnectorProperty("s3.endpoint",
hiveFlociDataLake.floci().endpoint().toString())
.addConnectorProperty("s3.path-style-access", "true");
}
@@ -111,6 +112,8 @@ public final class HudiQueryRunner
DistributedQueryRunner queryRunner = super.build();
try {
queryRunner.installPlugin(new
TestingHudiPlugin(queryRunner.getCoordinator().getBaseDataDir().resolve("hudi_data")));
+ queryRunner.installPlugin(new MemoryBlobCachePlugin());
+ queryRunner.loadBlobCacheManager("memory",
Map.of("fs.memory-cache.max-size", "128MB"));
queryRunner.createCatalog("hudi", "hudi", connectorProperties);
// Hudi connector does not support creating schema or any
other write operations
@@ -153,19 +156,19 @@ public final class HudiQueryRunner
}
}
- public static final class HudiMinioQueryRunnerMain
+ public static final class HudiFlociQueryRunnerMain
{
- private HudiMinioQueryRunnerMain() {}
+ private HudiFlociQueryRunnerMain() {}
public static void main(String[] args)
throws Exception
{
Logging.initialize();
- Logger log = Logger.get(HudiMinioQueryRunnerMain.class);
+ Logger log = Logger.get(HudiFlociQueryRunnerMain.class);
- Hive3MinioDataLake hiveMinioDataLake = new
Hive3MinioDataLake("test-bucket");
- hiveMinioDataLake.start();
- QueryRunner queryRunner = builder(hiveMinioDataLake)
+ Hive3FlociDataLake hiveFlociDataLake = new
Hive3FlociDataLake("test-bucket");
+ hiveFlociDataLake.start();
+ QueryRunner queryRunner = builder(hiveFlociDataLake)
.addCoordinatorProperty("http-server.http.port", "8080")
.setDataLoader(new
TpchHudiTablesInitializer(TpchTable.getTables()))
.build();
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
index 22f569dcf50d..e245b0e61396 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCacheFileOperations.java
@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableMultiset;
import com.google.common.collect.Multiset;
import io.airlift.units.Duration;
+import io.trino.blob.cache.alluxio.AlluxioBlobCachePlugin;
import io.trino.plugin.hudi.testing.ResourceHudiTablesInitializer;
import io.trino.plugin.hudi.util.FileOperationUtils.FileOperation;
import io.trino.testing.AbstractTestQueryFramework;
@@ -63,8 +64,6 @@ public class TestHudiAlluxioCacheFileOperations
Map<String, String> hudiProperties = ImmutableMap.<String,
String>builder()
.put("hudi.metadata-enabled", "true")
.put("fs.cache.enabled", "true")
- .put("fs.cache.directories",
cacheDirectory.toAbsolutePath().toString())
- .put("fs.cache.max-sizes", "100MB")
.put("hudi.metadata.cache.enabled", "false")
// Disable the async table-statistics refresh: on the first
query it reads the index
// definitions and table-property files (and the metadata
table) on a background
@@ -77,6 +76,11 @@ public class TestHudiAlluxioCacheFileOperations
return HudiQueryRunner.builder()
.addConnectorProperties(hudiProperties)
.setDataLoader(new ResourceHudiTablesInitializer())
+ .withPlugin(new AlluxioBlobCachePlugin())
+ .withBlobCache("alluxio", ImmutableMap.<String,
String>builder()
+ .put("fs.cache.directories",
cacheDirectory.toAbsolutePath().toString())
+ .put("fs.cache.max-sizes", "100MB")
+ .buildOrThrow())
.setWorkerCount(0)
.build();
}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
index 9c191a20e34c..0e5eea5d1df9 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiAlluxioCachingSmokeTest.java
@@ -21,6 +21,8 @@ import org.junit.jupiter.api.parallel.ResourceLock;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Map;
+import java.util.Optional;
import static com.google.common.io.MoreFiles.deleteRecursively;
import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
@@ -52,8 +54,15 @@ public class TestHudiAlluxioCachingSmokeTest
{
return ImmutableMap.<String, String>builder()
.put("fs.cache.enabled", "true")
+ .buildOrThrow();
+ }
+
+ @Override
+ protected Optional<Map<String, String>> getBlobCacheProperties()
+ {
+ return Optional.of(ImmutableMap.<String, String>builder()
.put("fs.cache.directories",
cacheDirectory.toAbsolutePath().toString())
.put("fs.cache.max-sizes", "1GB")
- .buildOrThrow();
+ .buildOrThrow());
}
}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMinioConnectorSmokeTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiFlociConnectorSmokeTest.java
similarity index 77%
rename from
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMinioConnectorSmokeTest.java
rename to
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiFlociConnectorSmokeTest.java
index 5fa1332073cd..b17b36001ce1 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMinioConnectorSmokeTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiFlociConnectorSmokeTest.java
@@ -13,7 +13,7 @@
*/
package io.trino.plugin.hudi;
-import io.trino.plugin.hive.containers.Hive3MinioDataLake;
+import io.trino.plugin.hive.containers.Hive3FlociDataLake;
import io.trino.plugin.hudi.testing.TpchHudiTablesInitializer;
import io.trino.testing.QueryRunner;
@@ -21,7 +21,7 @@ import static
io.trino.plugin.hive.containers.HiveHadoop.HIVE3_IMAGE;
import static io.trino.plugin.hudi.testing.HudiTestUtils.COLUMNS_TO_HIDE;
import static io.trino.testing.TestingNames.randomNameSuffix;
-public class TestHudiMinioConnectorSmokeTest
+public class TestHudiFlociConnectorSmokeTest
extends BaseHudiConnectorSmokeTest
{
@Override
@@ -29,11 +29,10 @@ public class TestHudiMinioConnectorSmokeTest
throws Exception
{
String bucketName = "test-hudi-connector-" + randomNameSuffix();
- Hive3MinioDataLake hiveMinioDataLake = closeAfterClass(new
Hive3MinioDataLake(bucketName, HIVE3_IMAGE));
- hiveMinioDataLake.start();
- hiveMinioDataLake.getMinioClient().ensureBucketExists(bucketName);
+ Hive3FlociDataLake hiveFlociDataLake = closeAfterClass(new
Hive3FlociDataLake(bucketName, HIVE3_IMAGE));
+ hiveFlociDataLake.start();
- return HudiQueryRunner.builder(hiveMinioDataLake)
+ return HudiQueryRunner.builder(hiveFlociDataLake)
.addConnectorProperty("hudi.columns-to-hide", COLUMNS_TO_HIDE)
.setDataLoader(new
TpchHudiTablesInitializer(REQUIRED_TPCH_TABLES))
.build();
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
index b5ada5741411..4ff361b8e6c1 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMemoryCacheFileOperations.java
@@ -76,7 +76,7 @@ public class TestHudiMemoryCacheFileOperations
assertFileSystemAccesses(
query,
ImmutableMultiset.<FileOperation>builder()
- .addCopies(new
FileOperation("FileSystemCache.cacheInput", DATA), 2)
+ .addCopies(new FileOperation("BlobCache.get", DATA), 2)
.addCopies(new FileOperation("InputFile.newStream",
INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream",
METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream",
TABLE_PROPERTIES), 2)
@@ -85,7 +85,7 @@ public class TestHudiMemoryCacheFileOperations
assertFileSystemAccesses(
query,
ImmutableMultiset.<FileOperation>builder()
- .addCopies(new
FileOperation("FileSystemCache.cacheInput", DATA), 2)
+ .addCopies(new FileOperation("BlobCache.get", DATA), 2)
.addCopies(new FileOperation("InputFile.newStream",
INDEX_DEFINITION), 2)
.add(new FileOperation("InputFile.newStream",
METADATA_TABLE_PROPERTIES))
.addCopies(new FileOperation("InputFile.newStream",
TABLE_PROPERTIES), 2)
@@ -102,7 +102,7 @@ public class TestHudiMemoryCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperation>builder()
- .addCopies(new
FileOperation("FileSystemCache.cacheInput", DATA), 6)
+ .addCopies(new FileOperation("BlobCache.get", DATA), 6)
.addCopies(new FileOperation("InputFile.newStream",
INDEX_DEFINITION), 4)
.addCopies(new FileOperation("InputFile.newStream",
METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperation("InputFile.newStream",
TABLE_PROPERTIES), 4)
@@ -110,7 +110,7 @@ public class TestHudiMemoryCacheFileOperations
assertFileSystemAccesses(query,
ImmutableMultiset.<FileOperation>builder()
- .addCopies(new
FileOperation("FileSystemCache.cacheInput", DATA), 6)
+ .addCopies(new FileOperation("BlobCache.get", DATA), 6)
.addCopies(new FileOperation("InputFile.newStream",
INDEX_DEFINITION), 4)
.addCopies(new FileOperation("InputFile.newStream",
METADATA_TABLE_PROPERTIES), 2)
.addCopies(new FileOperation("InputFile.newStream",
TABLE_PROPERTIES), 4)
@@ -127,7 +127,7 @@ public class TestHudiMemoryCacheFileOperations
private static Multiset<FileOperation> getFileOperations(QueryRunner
queryRunner)
{
return queryRunner.getSpans().stream()
- .filter(span -> span.getName().startsWith("Input.") ||
span.getName().startsWith("InputFile.") ||
span.getName().startsWith("FileSystemCache."))
+ .filter(span -> span.getName().startsWith("Input.") ||
span.getName().startsWith("InputFile.") ||
span.getName().startsWith("FileSystemCache.") ||
span.getName().startsWith("BlobCache."))
.filter(span ->
!span.getName().startsWith("InputFile.newInput"))
.filter(span -> !span.getName().startsWith("InputFile.exists"))
.filter(span ->
!isTrinoSchemaOrPermissions(getFileLocation(span)))
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
index 6f8971f3bbe6..51d0ae1a375a 100644
--- a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
+++ b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSmokeTest.java
@@ -17,6 +17,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Resources;
import io.trino.Session;
+import io.trino.blob.cache.alluxio.AlluxioBlobCachePlugin;
import io.trino.filesystem.Location;
import io.trino.filesystem.TrinoFileSystemFactory;
import io.trino.filesystem.TrinoInputFile;
@@ -100,10 +101,13 @@ public class TestHudiSmokeTest
protected QueryRunner createQueryRunner()
throws Exception
{
- return HudiQueryRunner.builder()
+ HudiQueryRunner.Builder builder = HudiQueryRunner.builder()
.setDataLoader(new ResourceHudiTablesInitializer())
- .addConnectorProperties(getAdditionalHudiProperties())
- .build();
+ .addConnectorProperties(getAdditionalHudiProperties());
+ getBlobCacheProperties().ifPresent(cacheProperties -> builder
+ .withPlugin(new AlluxioBlobCachePlugin())
+ .withBlobCache("alluxio", cacheProperties));
+ return builder.build();
}
protected ImmutableMap<String, String> getAdditionalHudiProperties()
@@ -111,6 +115,11 @@ public class TestHudiSmokeTest
return ImmutableMap.of();
}
+ protected Optional<Map<String, String>> getBlobCacheProperties()
+ {
+ return Optional.empty();
+ }
+
@Test
public void testReadNonPartitionedTable()
{
@@ -928,6 +937,42 @@ public class TestHudiSmokeTest
assertQuery(query, "VALUES (1, 'a1', 100.0, 1000), (3, 'a3', 101.0,
1001)");
}
+ @ParameterizedTest
+ @EnumSource(
+ value = ResourceHudiTablesInitializer.TestingTable.class,
+ names = {"HUDI_MULTI_FG_PT_V6_MOR", "HUDI_MULTI_FG_PT_V8_MOR"})
+ public void
testDynamicFilterEliminatesAllSplits(ResourceHudiTablesInitializer.TestingTable
table)
+ {
+ Session session = SessionBuilder
+ .from(getSession())
+ .withDynamicFilterTimeout("10s")
+ .build();
+ final String tableIdentifier = "hudi:tests." + table.getRoTableName();
+
+ // The build side matches no rows, so the completed dynamic filter is
NONE and the
+ // probe-side split source must report itself finished instead of
draining the queue
+ @Language("SQL") String query = "SELECT t1.id FROM " +
+ table + " t1 " +
+ "INNER JOIN " + table + " t2 ON t1.id = t2.id " +
+ "WHERE t2.price < 0";
+ MaterializedResult explainRes = getQueryRunner().execute(session,
"EXPLAIN ANALYZE " + query);
+ Pattern scanFilterInputRowsPattern =
getScanFilterInputRowsPattern(tableIdentifier);
+ Matcher matcher =
scanFilterInputRowsPattern.matcher(explainRes.toString());
+ assertThat(matcher.find())
+ .withFailMessage("Could not find 'ScanFilter' for table '%s'
with 'dynamicFilters' and 'Input: X rows' stats in EXPLAIN output.\nOutput
was:\n%s",
+ tableIdentifier, explainRes.toString())
+ .isTrue();
+
+ // Zero probe-side input rows pins split elimination at the source:
without the NONE
+ // short-circuit the probe would scan all rows and the join would
discard them, which
+ // returns the same empty result but reads Input: 4 rows here
+ assertThat(Long.parseLong(matcher.group(1)))
+ .describedAs("Probe side (%s) should read no rows when the
dynamic filter is NONE", tableIdentifier)
+ .isEqualTo(0);
+
+ assertThat(getQueryRunner().execute(session,
query).getRowCount()).isEqualTo(0);
+ }
+
@ParameterizedTest
@EnumSource(
value = ResourceHudiTablesInitializer.TestingTable.class,
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSplitSource.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSplitSource.java
new file mode 100644
index 000000000000..567d4b776459
--- /dev/null
+++ b/hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiSplitSource.java
@@ -0,0 +1,121 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.hudi;
+
+import io.airlift.units.Duration;
+import io.trino.plugin.hive.util.AsyncQueue;
+import io.trino.plugin.hive.util.ThrottledAsyncQueue;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ConnectorSplit;
+import io.trino.spi.connector.DynamicFilterSnapshot;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.predicate.TupleDomain;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class TestHudiSplitSource
+{
+ private static final SchemaTableName TABLE = new SchemaTableName("tests",
"split_source_test");
+ private static final Duration TIMEOUT = new Duration(10, SECONDS);
+ private static final ConnectorSplit DUMMY_SPLIT = new ConnectorSplit() {};
+
+ private final ExecutorService executor = Executors.newCachedThreadPool();
+ private final ScheduledExecutorService scheduler =
Executors.newSingleThreadScheduledExecutor();
+
+ @AfterAll
+ public void tearDown()
+ {
+ executor.shutdownNow();
+ scheduler.shutdownNow();
+ }
+
+ @Test
+ public void testNoneDynamicFilterTerminatesSource()
+ throws Exception
+ {
+ AsyncQueue<ConnectorSplit> queue = new ThrottledAsyncQueue<>(100, 100,
executor);
+ // Loader that stays "in flight": splits are queued but the queue is
never finished,
+ // so only the early-termination path can make the source report
finished
+ HudiSplitSource splitSource = new HudiSplitSource(queue, scheduler,
(q, errorListener) -> () -> q.offer(DUMMY_SPLIT), TABLE, TIMEOUT);
+
+ CompletableFuture<List<ConnectorSplit>> batch =
+ splitSource.getNextBatch(10, new
DynamicFilterSnapshot(TupleDomain.none(), true));
+
+ assertThat(batch).isCompletedWithValueMatching(List::isEmpty);
+ assertThat(splitSource.isFinished()).isTrue();
+ }
+
+ @Test
+ public void testLoaderFailureBlocksFinishedAndSurfaces()
+ throws Exception
+ {
+ AsyncQueue<ConnectorSplit> queue = new ThrottledAsyncQueue<>(100, 100,
executor);
+ HudiSplitSource splitSource = new HudiSplitSource(
+ queue,
+ scheduler,
+ (q, errorListener) -> () -> errorListener.accept(new
RuntimeException("boom")),
+ TABLE,
+ TIMEOUT);
+ awaitQueueFinished(queue);
+
+ // The failed load finished the queue, but the source must not claim
finished while the
+ // failure is undelivered; the next batch is what surfaces it
+ assertThat(splitSource.isFinished()).isFalse();
+ CompletableFuture<List<ConnectorSplit>> batch =
+ splitSource.getNextBatch(10, new
DynamicFilterSnapshot(TupleDomain.all(), false));
+ assertThatThrownBy(batch::join)
+ .hasCauseInstanceOf(TrinoException.class)
+ .hasMessageContaining("Failed to generate splits");
+ }
+
+ @Test
+ public void testCompletedLoaderFinishesNormally()
+ throws Exception
+ {
+ AsyncQueue<ConnectorSplit> queue = new ThrottledAsyncQueue<>(100, 100,
executor);
+ HudiSplitSource splitSource = new HudiSplitSource(
+ queue,
+ scheduler,
+ (q, errorListener) -> (Runnable) q::finish,
+ TABLE,
+ TIMEOUT);
+ awaitQueueFinished(queue);
+
+ CompletableFuture<List<ConnectorSplit>> batch =
+ splitSource.getNextBatch(10, new
DynamicFilterSnapshot(TupleDomain.all(), false));
+ assertThat(batch.join()).isEmpty();
+ assertThat(splitSource.isFinished()).isTrue();
+ }
+
+ private static void awaitQueueFinished(AsyncQueue<ConnectorSplit> queue)
+ throws InterruptedException
+ {
+ for (int i = 0; i < 500 && !queue.isFinished(); i++) {
+ Thread.sleep(10);
+ }
+ assertThat(queue.isFinished()).isTrue();
+ }
+}
diff --git
a/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestTupleDomainUtilsTest.java
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestTupleDomainUtilsTest.java
index 64becba2ce60..e6e4557580ca 100644
---
a/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestTupleDomainUtilsTest.java
+++
b/hudi-trino/src/test/java/io/trino/plugin/hudi/util/TestTupleDomainUtilsTest.java
@@ -176,7 +176,7 @@ class TestTupleDomainUtilsTest
// "other_col" is an irrelevant column
TupleDomain<String> tupleDomain = TupleDomain.withColumnDomains(Map.of(
"key1", Domain.singleValue(BIGINT, 1L), // EQUALS
- "key2", Domain.multipleValues(VARCHAR, List.of("a", "b")), //
IN
+ "key2", Domain.multipleValues(VARCHAR,
List.of(Slices.utf8Slice("a"), Slices.utf8Slice("b"))), // IN
"other_col",
Domain.create(ValueSet.ofRanges(Range.greaterThan(BIGINT, 10L)), false)));
List<String> sourceFields = List.of("key1", "key2");
assertThat(TupleDomainUtils.areDomainsInOrEqualOnly(tupleDomain,
sourceFields)).isTrue();
diff --git a/pom.xml b/pom.xml
index f64fde294508..ec7c9c7cafe9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -131,7 +131,14 @@
<hive.parquet.version>1.10.1</hive.parquet.version>
<hive.avro.version>1.11.4</hive.avro.version>
<presto.version>0.273</presto.version>
- <trino.version>481</trino.version>
+ <!-- trino.sha is the exact trinodb/trino commit hudi-trino builds
against; its project
+ version must equal trino.version and the two advance only together.
trino.e2e.version is
+ the latest released Trino (e2e server image, trino-jdbc for the integ
tests); the nightly
+ pin-advance job refreshes all three, humans merge its bot/trino-pin
branch. Keep each
+ property on one line: the workflows and
scripts/trino/bootstrap_trino.sh read them with sed. -->
+ <trino.version>484-SNAPSHOT</trino.version>
+ <trino.sha>5b82ec9e7116ec1ed3a83f4cc2f8cf9aaa87b12f</trino.sha>
+ <trino.e2e.version>483</trino.e2e.version>
<hive.exec.classifier>core</hive.exec.classifier>
<metrics.version>4.1.1</metrics.version>
<orc.spark.version>1.6.0</orc.spark.version>
@@ -1758,7 +1765,7 @@
<dependency>
<groupId>io.trino</groupId>
<artifactId>trino-jdbc</artifactId>
- <version>${trino.version}</version>
+ <version>${trino.e2e.version}</version>
</dependency>
<!-- Zookeeper -->
diff --git a/release/release_guide.md b/release/release_guide.md
index ed78098f66c1..3175b63b96c6 100644
--- a/release/release_guide.md
+++ b/release/release_guide.md
@@ -290,6 +290,31 @@ Here is how to go about a bug fix release.
- Go to apache/hudi repo locally and pull this branch. Here after you can work
on this branch and push to origin when need be.
- Do not forget to set the env variables from above section.
+## hudi-trino Trino pin-back
+
+On master hudi-trino tracks `trinodb/trino` master at the commit in
`trino.sha`, whose `trino.version` is a
+`-SNAPSHOT` that resolves from nowhere but a local build. A release must
depend on a released Trino, and the pin-back
+must land on the release branch before the source release is generated (see
"Build a release candidate", the Generate
+Source Release step) -- otherwise the voted tarball ships a `-SNAPSHOT` Trino
pin that cannot be built from Central.
+
+1. Wait for the latest released Trino `NNN` to be available on Maven Central.
+2. In a `trinodb/trino` checkout, find the tagged commit: `TAG_SHA=$(git
rev-list -n1 NNN)`.
+3. If the pin is behind the tag, advance master's pin to `TAG_SHA` first by
dispatching the
+ `Hudi Trino SPI Compatibility` workflow with `trino_ref=NNN` (it then
verifies and pins exactly that tag rather
+ than master HEAD) and merging the pin PR a committer opens from the pushed
`bot/trino-pin` branch. If the pin is ahead of the tag, enumerate the
adaptations that would be lost with
+ `git log NNN..<pin> -- core/trino-spi lib/trino-filesystem
lib/trino-filesystem-manager lib/trino-hdfs lib/trino-memory-context`
+ and revert them forward on the release branch only, never on master.
+4. On the release branch set `trino.version=NNN`, `trino.sha=TAG_SHA` and
`trino.e2e.version=NNN` in the root
+ pom, the `<parent>` version in `docker/trino/shim/pom.xml`, and the `ARG
TRINO_VERSION` default in
+ `docker/trino/Dockerfile` (`build_image.sh` reads `trino.e2e.version` from
the root pom). Re-check SPI-surface-coupled
+ dependency scopes against `NNN` (e.g. `jts-core` is `provided` because it
joined the Trino SPI surface in 482;
+ the shim's SpiDependencyChecker fails the build loudly if a scope no longer
matches the target release).
+5. Verify the released Trino resolves from Central against an empty local
repository
+ (scope the check to io.trino: the module's hudi siblings are not on Central
until this release completes):
+ `mvn dependency:get -Dartifact=io.trino:trino-hive:NNN
-Dmaven.repo.local=$(mktemp -d)`
+6. CI and the E2E workflow then run with zero SPI drift; the staging deploy
flow in "Build a release candidate"
+ is unchanged.
+
## Verify that a Release Build Works
Run "mvn -Prelease clean install" to ensure that the build processes are in
good shape. // You need to execute this command once you have the release
branch in apache/hudi
@@ -429,6 +454,8 @@ Set up a few environment variables to simplify Maven
commands that follow. This
and `./scripts/release/deploy_staging_jars_java25.sh 2>&1 | tee -a
"/tmp/${RELEASE_VERSION}-${RC_NUM}.deploy3.log"`.
This step must run after the Java 11 step in 9.4.1, which installs
the upstream Hudi modules that hudi-trino
resolves from the local m2 (the script does not pass `-am` because
Lombok cannot run on JDK 25).
+ 4. The hudi-trino Trino pin-back already happened when the release
branch was cut (see the
+ "hudi-trino Trino pin-back" section under "Cut a release branch");
the Java 25 deploy needs no extra steps.
5. Note that each of the Java 17 and Java 25 builds uploads its artifacts
to its own separate staging repo. Use the
`copy_staging_repo.sh` script once per extra staging repo to copy all
artifacts into the Java 11 staging repo
so that all artifacts stay in the same repo.
diff --git a/rfc/rfc-105/rfc-105.md b/rfc/rfc-105/rfc-105.md
index 51c7aef2fb90..dce625a08cb3 100644
--- a/rfc/rfc-105/rfc-105.md
+++ b/rfc/rfc-105/rfc-105.md
@@ -125,7 +125,7 @@ apache/hudi : hudi-trino-plugin/ (Maven profile
-Phudi-trino,
| `src/main/java/io/trino/plugin/hudi/HudiPlugin.java` | Implements
`io.trino.spi.Plugin`. Single method returning `new HudiConnectorFactory()`
(from the `hudi-trino` artifact). ~10 lines. |
| `src/main/resources/META-INF/services/io.trino.spi.Plugin` | Service-loader
pointer to `io.trino.plugin.hudi.HudiPlugin`. |
| `pom.xml` | `<packaging>trino-plugin</packaging>`; pins
`org.apache.hudi:hudi-trino:<version>`; SPI deps as `provided`. |
-| `src/test/java/...` | All current Trino-side tests stay: `HudiQueryRunner`,
`TestHudiSmokeTest`, `TestHudiMinioConnectorSmokeTest`,
`TestHudiConnectorTest`, `TestHudiSharedMetastore`, `TestHudiSystemTables`,
`TestHudiPlugin`, `TestHudiConfig`, plus data initializers. Required by the
Trino-side test-coverage commitment. |
+| `src/test/java/...` | All current Trino-side tests stay: `HudiQueryRunner`,
`TestHudiSmokeTest`, `TestHudiFlociConnectorSmokeTest`,
`TestHudiConnectorTest`, `TestHudiSharedMetastore`, `TestHudiSystemTables`,
`TestHudiPlugin`, `TestHudiConfig`, plus data initializers. Required by the
Trino-side test-coverage commitment. |
#### Hudi-side `hudi-trino-plugin/` (the engine)
@@ -187,7 +187,7 @@ On the Trino side, existing CI continues to build and test
`plugin/trino-hudi`,
### Test strategy
-**Full test duplication.** The Trino-side smoke tests (`TestHudiSmokeTest`,
`TestHudiMinioConnectorSmokeTest`, `TestHudiConnectorTest`, etc.) are mirrored
on the Hudi side and additionally extended.
+**Full test duplication.** The Trino-side smoke tests (`TestHudiSmokeTest`,
`TestHudiFlociConnectorSmokeTest`, `TestHudiConnectorTest`, etc.) are mirrored
on the Hudi side and additionally extended.
- **Trino side runs them** on every Trino PR — fulfilling the Trino-side
test-coverage commitment.
- **Hudi side runs them** on every Hudi PR touching `hudi-trino-plugin` — so
Hudi contributors catch regressions before they ship in a Hudi release. The
Hudi-side suite is also **expanded** with more granular unit tests covering
split generation edge cases, all eight index-support strategies, the MOR
record-merging path, lazy-commit-time snapshot isolation, and the cache-key
provider.
diff --git a/scripts/trino/bootstrap_trino.sh b/scripts/trino/bootstrap_trino.sh
new file mode 100755
index 000000000000..835c6389145d
--- /dev/null
+++ b/scripts/trino/bootstrap_trino.sh
@@ -0,0 +1,170 @@
+#!/bin/bash
+
+#
+# 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.
+#
+
+##
+## Builds the trinodb/trino modules that hudi-trino compiles and tests against
+## and installs them into the local Maven repository.
+##
+## On master, hudi-trino tracks trinodb/trino master at the commit pinned by
the
+## <trino.sha> property in the root pom (whose project version is
<trino.version>,
+## e.g. 484-SNAPSHOT). Trino publishes no SNAPSHOT artifacts, so every io.trino
+## dependency must be installed from source by this script before hudi-trino
can
+## build. On release branches <trino.version> is a released number and only the
+## test-jars (never published by Trino) need this script.
+##
+## Usage: bootstrap_trino.sh <path-to-trino-checkout> [--ref <sha>]
[--skip-checkout] [--keep-m2]
+##
+## <path-to-trino-checkout> a trinodb/trino git checkout (clean worktree)
+## --ref <sha> build at this commit instead of the pinned
<trino.sha>
+## (used by the nightly pin-advance workflow at
trino HEAD;
+## relaxes the version cross-check to a warning)
+## --skip-checkout assume the worktree is already at the right
commit
+## (CI checks out the SHA itself via
actions/checkout)
+## --keep-m2 do not purge ~/.m2/repository/io/trino first
+## (the purge guards against stale artifacts under
the
+## same SNAPSHOT coordinates from an older pin)
+##
+
+set -euo pipefail
+
+HUDI_ROOT="$(cd "$(dirname "$0")/../.." && pwd)"
+
+TRINO_REPO=""
+REF=""
+SKIP_CHECKOUT="false"
+KEEP_M2="false"
+
+while [[ $# -gt 0 ]]; do
+ case "$1" in
+ --ref)
+ REF="$2"
+ shift 2
+ ;;
+ --skip-checkout)
+ SKIP_CHECKOUT="true"
+ shift
+ ;;
+ --keep-m2)
+ KEEP_M2="true"
+ shift
+ ;;
+ -h|--help)
+ grep '^##' "$0" | sed 's/^## \{0,1\}//'
+ exit 0
+ ;;
+ *)
+ if [[ -z "$TRINO_REPO" ]]; then
+ TRINO_REPO="$1"
+ shift
+ else
+ echo "ERROR: unexpected argument: $1" >&2
+ exit 1
+ fi
+ ;;
+ esac
+done
+
+if [[ -z "$TRINO_REPO" || ! -d "$TRINO_REPO/.git" && ! -f "$TRINO_REPO/.git"
]]; then
+ echo "ERROR: first argument must be a trinodb/trino git checkout" >&2
+ echo "Usage: $0 <path-to-trino-checkout> [--ref <sha>] [--skip-checkout]
[--keep-m2]" >&2
+ exit 1
+fi
+
+# The pin lives in the root pom; keep each property on one line (this sed and
the
+# CI workflows depend on it).
+PINNED_SHA=$(sed -n 's|.*<trino.sha>\(.*\)</trino.sha>.*|\1|p'
"$HUDI_ROOT/pom.xml")
+PINNED_VERSION=$(sed -n 's|.*<trino.version>\(.*\)</trino.version>.*|\1|p'
"$HUDI_ROOT/pom.xml")
+
+if [[ -z "$PINNED_VERSION" ]]; then
+ echo "ERROR: could not read <trino.version> from $HUDI_ROOT/pom.xml" >&2
+ exit 1
+fi
+
+RELAXED_VERSION_CHECK="false"
+if [[ -n "$REF" ]]; then
+ RELAXED_VERSION_CHECK="true"
+else
+ REF="$PINNED_SHA"
+ if [[ -z "$REF" ]]; then
+ echo "ERROR: could not read <trino.sha> from $HUDI_ROOT/pom.xml and no
--ref given" >&2
+ exit 1
+ fi
+fi
+
+# JDK gate: trino at head enforces JDK 25.
+JAVA_MAJOR=$(java -version 2>&1 | awk -F[\".] '/version/ {print $2}')
+if [[ "$JAVA_MAJOR" != "25" ]]; then
+ echo "ERROR: JDK 25 required to build trino (found major version:
${JAVA_MAJOR:-unknown})." >&2
+ echo "Hint: export JAVA_HOME=\$(/usr/libexec/java_home -v 25)" >&2
+ exit 1
+fi
+
+if ! git -C "$TRINO_REPO" rev-parse --verify --quiet "${REF}^{commit}" >
/dev/null; then
+ echo "ERROR: commit $REF not present in $TRINO_REPO" >&2
+ echo "Hint: git -C $TRINO_REPO fetch https://github.com/trinodb/trino.git
master --tags" >&2
+ exit 1
+fi
+
+if [[ "$SKIP_CHECKOUT" == "false" ]]; then
+ # Refuse tracked modifications; untracked files are harmless for a checkout.
+ if [[ -n "$(git -C "$TRINO_REPO" status --porcelain -uno)" ]]; then
+ echo "ERROR: $TRINO_REPO has uncommitted tracked changes; commit or stash
them first" >&2
+ exit 1
+ fi
+ git -C "$TRINO_REPO" -c advice.detachedHead=false checkout --detach "$REF"
+fi
+
+ACTUAL_VERSION=$("$TRINO_REPO/mvnw" -q -N -f "$TRINO_REPO/pom.xml"
help:evaluate -Dexpression=project.version -DforceStdout)
+if [[ "$ACTUAL_VERSION" != "$PINNED_VERSION" ]]; then
+ if [[ "$RELAXED_VERSION_CHECK" == "true" ]]; then
+ echo "WARNING: trino at $REF has version $ACTUAL_VERSION but the pom pins
<trino.version>$PINNED_VERSION</trino.version> (version rollover?)"
+ else
+ echo "ERROR: trino at pinned sha $REF has version $ACTUAL_VERSION, but the
root pom says <trino.version>$PINNED_VERSION</trino.version>." >&2
+ echo "The pin properties must advance together; fix the pom or your
checkout." >&2
+ exit 1
+ fi
+fi
+
+if [[ "$KEEP_M2" == "false" ]]; then
+ # SNAPSHOT coordinates do not change when the pin does, so artifacts from an
+ # older pin are indistinguishable from current ones. Purge to be safe.
+ echo "Purging ~/.m2/repository/io/trino (use --keep-m2 to skip)"
+ rm -rf "$HOME/.m2/repository/io/trino"
+fi
+
+export MAVEN_OPTS="${MAVEN_OPTS:--Xmx4g}"
+
+# Module list rationale:
+# - trino-hive, trino-filesystem-manager, trino-parquet, trino-plugin-toolkit
and
+# their -am closure cover every compile-scope io.trino dependency of
hudi-trino
+# (trino-spi, trino-cache, trino-filesystem, trino-hive-formats,
+# trino-memory-context, trino-metastore come in transitively).
+# - The remaining modules are hudi-trino's test-scope dependencies.
+# - trino-spi, trino-filesystem, trino-hive, trino-main are the four test-jar
+# producers needed by the hudi-trino-tests profile; -DskipTests (NOT
+# -Dmaven.test.skip) is load-bearing: test-jars need compiled test classes.
+# - The -am closure also installs the trino-root pom, which the hudi-trino BOM
+# import and the docker/trino/shim parent resolve.
+# - The blob-cache plugins back the cache managers the hudi-trino tests load
+# (memory unconditionally via HudiQueryRunner, alluxio in the caching tests).
+echo "Building trino modules at $REF (version $ACTUAL_VERSION); this takes
roughly 10-30 minutes"
+(cd "$TRINO_REPO" && ./mvnw install -am -DskipTests -Dair.check.skip-all=true
-T1C \
+ -pl
:trino-hive,:trino-filesystem-manager,:trino-parquet,:trino-plugin-toolkit,:trino-main,:trino-testing,:trino-testing-containers,:trino-testing-services,:trino-client,:trino-parser,:trino-hdfs,:trino-tpch,:trino-spi,:trino-filesystem,:trino-blob-cache-memory,:trino-blob-cache-alluxio)
+
+echo "Done. io.trino artifacts for $ACTUAL_VERSION are installed in ~/.m2."
