You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/10/01 16:27:12 UTC
Re: [PR] [trino] Add initial connector build and plugin skeleton [fluss]
fresh-borzoni commented on code in PR #4207:
URL: https://github.com/apache/fluss/pull/4207#discussion_r4157746817
##########
.github/workflows/ci-trino.yaml:
##########
@@ -0,0 +1,63 @@
+################################################################################
+# 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.
+################################################################################
+
+name: Trino Connector CI
+
+permissions:
+ contents: read
+
+on:
+ push:
+ branches:
+ - main
+ - release-**
+ paths:
+ - 'fluss-trino/**'
+ - 'tools/maven/checkstyle.xml'
+ - 'tools/maven/suppressions.xml'
+ - '.github/workflows/ci-trino.yaml'
+
+ pull_request:
+ paths:
+ - 'fluss-trino/**'
+ - 'tools/maven/checkstyle.xml'
+ - 'tools/maven/suppressions.xml'
+ - '.github/workflows/ci-trino.yaml'
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
+ cancel-in-progress: true
+
+jobs:
+ build:
+ name: "Java 25 / Trino"
+ runs-on: ubuntu-latest
+
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v6
+
+ - name: Set up JDK 25
+ uses: actions/setup-java@v5
+ with:
+ distribution: 'temurin'
+ java-version: '25'
+
+ - name: Build
+ working-directory: fluss-trino
+ run: ./mvnw -B clean verify
Review Comment:
This job will fail at dependency resolution. fluss-trino is a separate Maven build, so it looks for fluss-client:1.1-SNAPSHOT in Maven repos instead of building it from the checkout.
On a fresh runner it fails with Could not find artifact org.apache.fluss:fluss-client:jar:1.1-SNAPSHOT, and the same for fluss-server, fluss-test-utils and the test jars. client-integration.yml already solves this by building Fluss from the checkout on JDK 17 first.
Shall we do the same here: install fluss-client, fluss-server and fluss-test-utils with -am, then build the connector on JDK 25?
##########
fluss-trino/src/main/java/org/apache/fluss/trino/FlussLogSplitReader.java:
##########
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.trino;
+
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.ScanRecords;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.shaded.guava32.com.google.common.collect.ImmutableList;
+
+import io.trino.spi.TrinoException;
+
+import java.time.Duration;
+import java.util.List;
+
+import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
+import static org.apache.fluss.trino.FlussErrorCode.FLUSS_READ_ERROR;
+import static org.apache.fluss.utils.ExceptionUtils.firstOrSuppressed;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Bounded log reader for one physical Fluss bucket.
+ *
+ * <p>The underlying {@link LogScanner} is unbounded. This reader restricts it to the split's
+ * exclusive-end offset range and reports {@link PollResult#FINISHED} once that range is consumed.
+ */
+final class FlussLogSplitReader implements FlussSplitReader {
+
+ private static final long ESTIMATED_RECORD_OVERHEAD_BYTES = 128;
+
+ private final FlussBucketHandle bucket;
+ private final TableBucket tableBucket;
+ private final FlussLogRange range;
+
+ private final LogScanner scanner;
+
+ private List<ScanRecord> records = ImmutableList.of();
+ private int recordIndex;
+ private int recordLimit;
+
+ private long estimatedRetainedBytes;
+ private long completedBytes;
+
+ private boolean stopReached;
+ private boolean finished;
+ private boolean closed;
+
+ FlussLogSplitReader(Table table, FlussBucketHandle bucket, FlussLogRange range) {
+ checkNotNull(table, "table is null");
+ this.bucket = checkNotNull(bucket, "bucket is null");
+ this.range = checkNotNull(range, "range is null");
+ this.tableBucket = bucket.toTableBucket();
+
+ if (range.isEmpty()) {
+ scanner = null;
+ finished = true;
+ return;
+ }
+
+ LogScanner scanner = table.newScan().createLogScanner();
+
+ try {
+ if (bucket.isPartitioned()) {
+ scanner.subscribe(
+ bucket.getRequiredPartitionId(),
+ bucket.getBucketId(),
+ range.getStartOffset());
+ } else {
+ scanner.subscribe(bucket.getBucketId(), range.getStartOffset());
+ }
+ } catch (RuntimeException | Error failure) {
+ try {
+ scanner.close();
+ } catch (Throwable closeFailure) {
+ firstOrSuppressed(closeFailure, failure);
+ }
+ throw failure;
+ }
+
+ this.scanner = scanner;
+ }
+
+ @Override
+ public PollResult poll(Duration timeout) {
+ checkNotNull(timeout, "timeout is null");
+
+ if (closed || finished) {
+ return PollResult.FINISHED;
+ }
+
+ if (hasNext()) {
+ throw new IllegalStateException(
+ "Cannot poll Fluss log scanner while the current batch is not consumed");
+ }
+
+ ScanRecords batch;
+ try {
+ batch = scanner.poll(timeout);
+ } catch (RuntimeException e) {
+ throw new TrinoException(FLUSS_READ_ERROR, "Failed reading Fluss log for " + bucket, e);
+ }
+ List<ScanRecord> batchRecords = batch.records(tableBucket);
+
+ long retainedBytes = 0;
+ int validRecordCount = 0;
+ boolean reachesStop = false;
+
+ for (ScanRecord record : batchRecords) {
+ long recordBytes = Math.max(0, record.getSizeInBytes());
+
+ // This is an estimate. LogScanner does not expose all client-side retained buffers.
+ retainedBytes += ESTIMATED_RECORD_OVERHEAD_BYTES + recordBytes;
+ long offset = record.logOffset();
+ if (offset < range.getStartOffset()) {
+ throw new TrinoException(
+ GENERIC_INTERNAL_ERROR,
+ "Fluss offset invariant failed for "
+ + bucket
+ + ": offset "
+ + offset
+ + " precedes "
+ + range);
+ }
+
+ /*
Review Comment:
Flink and Spark already have their own stopping-offset logic, and this is a third copy.
On dev@, @polyzos suggested moving it into fluss-client. I wonder if we can do it or there is any issue with this?
##########
fluss-trino/README.md:
##########
@@ -0,0 +1,180 @@
+<!--
+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.
+-->
+
+# Fluss Trino connector
+
+This read-only connector targets **Trino 483 and JDK 25**. It is built as a
+standalone Maven module. Source conventions follow Fluss, including Java 8 syntax
+and collection idioms; the resulting plugin requires JDK 25 and is not a Java 8
+binary.
+
+## Supported operations
+
+| Capability | Support |
+| --- | --- |
+| Database, table and column discovery | Supported |
+| `DESCRIBE`, `SHOW CREATE TABLE`, `"table$columns"` | Supported |
+| Non-partitioned and partitioned Log Table reads | Supported, with bounded offset ranges |
+| Non-partitioned and partitioned Primary Key Table reads | Supported, using bucket snapshots |
+| Lakehouse table reads | Not supported; metadata remains available |
+| Writes, DDL, lookup and Union Read | Not supported |
+| Predicate, aggregation, limit and partition pruning pushdown | Not implemented |
+| Task retries / fault-tolerant execution | Not supported |
+
+Ordinary `SELECT`, column selection and ordering, `COUNT(*)`, `WHERE`, `ORDER BY`
+and `LIMIT` use Trino's execution engine. Filters and other operations without
+pushdown are evaluated by Trino. Partitioned scans enumerate physical buckets;
+a partition predicate does not currently avoid scanning other partitions.
+
+Trino-visible database and table names are normalized to lowercase. When multiple
+Fluss names differ only by case, resolving the ambiguous name fails rather than
+arbitrarily choosing a physical table. Case-colliding column names are rejected.
+
+Supported values include BOOLEAN, integer types, FLOAT/DOUBLE, CHAR/STRING,
+BINARY/BYTES, DECIMAL, DATE/TIME, TIMESTAMP, TIMESTAMP_LTZ, and recursively
+ARRAY/MAP/ROW, subject to Trino's map-key type constraints. Decimal conversion
+preserves decimal precision. TIMESTAMP_LTZ preserves the instant using UTC;
+TIMESTAMP has no timezone conversion. Fluss TIME values contain milliseconds of
+the day, even when their schema declares a higher precision. Null map keys and
+keys that become duplicates under Trino semantics are rejected.
+
+## Build
+
+Build the connector with its own Maven wrapper and JDK 25:
+
+```sh
+cd fluss-trino
+JAVA_HOME=/path/to/jdk-25 ./mvnw clean package -DskipTests
+```
+
+## Install and configure
+
+Copy the contents of the assembled `target/fluss-trino-<version>/` directory,
+including its dependency JARs, into a dedicated `plugin/fluss/` directory on every
+Trino coordinator and worker. Use the assembled plugin, not just the connector
+JAR, and deploy the same version on all nodes.
+
+Create `etc/catalog/fluss.properties` on the Trino nodes:
+
+```properties
+connector.name=fluss
+bootstrap.servers=localhost:9123
+```
+
+Replace the bootstrap address with Fluss endpoints reachable from every Trino
+node. The connector also accepts `client.security.protocol`,
Review Comment:
SASL setup doesn't seem to work on JDK 25. A catalog with correct credentials fails with getSubject is not supported. The cause is SaslClientCallbackHandler in fluss-common, which calls Subject.getSubject(), and JDK 24+ no longer supports that call.
At least Fluss's own FlussAuthorizationITCase passes on JDK 17 and fails on JDK 25 when I tried to run it
##########
fluss-trino/src/main/java/org/apache/fluss/trino/FlussClientManager.java:
##########
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.trino;
+
+import com.google.inject.Inject;
+import jakarta.annotation.PreDestroy;
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.ConnectionFactory;
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.table.Table;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.TablePath;
+
+import static org.apache.fluss.utils.ExceptionUtils.firstOrSuppressed;
+import static org.apache.fluss.utils.ExceptionUtils.rethrowException;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/** Owns the shared Fluss connection and Admin client. */
+public final class FlussClientManager {
+ private static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
+ private static final String SECURITY_PROTOCOL = "client.security.protocol";
+ private static final String SASL_MECHANISM = "client.security.sasl.mechanism";
+ private static final String SASL_USERNAME = "client.security.sasl.username";
+ private static final String SASL_PASSWORD = "client.security.sasl.password";
+
+ private final Connection connection;
+ private final Admin admin;
+
+ @Inject
+ public FlussClientManager(FlussConfig config) {
+ checkNotNull(config, "config is null");
+
+ Configuration configuration = new Configuration();
+ configuration.setString(BOOTSTRAP_SERVERS, config.getBootstrapServers());
+
+ config.getSecurityProtocol()
+ .ifPresent(value -> configuration.setString(SECURITY_PROTOCOL, value));
+ config.getSaslMechanism()
+ .ifPresent(value -> configuration.setString(SASL_MECHANISM, value));
+ config.getSaslUsername().ifPresent(value -> configuration.setString(SASL_USERNAME, value));
+ config.getSaslPassword().ifPresent(value -> configuration.setString(SASL_PASSWORD, value));
+
+ connection = ConnectionFactory.createConnection(configuration);
Review Comment:
If Fluss is down when Trino starts, Trino doesn't start.
While Trino is running, the connector already recovers from a Fluss outage, so only startup is the problem. Could we create the connection lazily, on first use?
##########
.github/workflows/ci-trino.yaml:
##########
@@ -0,0 +1,63 @@
+################################################################################
+# 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.
+################################################################################
+
+name: Trino Connector CI
+
+permissions:
+ contents: read
+
+on:
+ push:
+ branches:
+ - main
+ - release-**
+ paths:
+ - 'fluss-trino/**'
Review Comment:
This job runs only when fluss-trino/ ** changes, so a fluss-client change that breaks the connector never runs it. Should we add fluss-client/ ** , fluss-common/ ** and fluss-rpc/ ** ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org